Disconnect after leaving the group

[abicky]

This PR removes an unnecessary reconnection after Kafka::Consumer#stop is called.

@cluster.disconnect closes the connections to all brokers including the coordinator in spite of the fact that @group.leave requires a connection to the coordinator, so call @cluster.disconnect after @group.leave.

You can confirm the behavior by executing the following script:

require "kafka"

CLIENT_ID = ENV["CLIENT_ID"]
GROUP_ID = ENV["GROUP_ID"]
TOPIC = ENV["TOPIC"]

$stdout.sync = true
logger = Logger.new($stdout)
logger.level = Logger::DEBUG
kafka = Kafka.new(["localhost:9092"], client_id: CLIENT_ID, logger: logger)

consumer = kafka.consumer(group_id: GROUP_ID)
consumer.subscribe(TOPIC)
Thread.new do
  # Wait for rebalance to end
  sleep 5
  consumer.stop
end

consumer.each_message do |message|
  puts message.topic, message.partition
  puts message.offset, message.key, message.value
end

Before

-- snip --
D, [2020-03-25T05:35:02.636289 #32996] DEBUG -- : [[g] {test: 0, 2, 1}:] Handling fetcher command: stop
I, [2020-03-25T05:35:02.636329 #32996]  INFO -- : [g] {test: 0, 2, 1}: Fetcher thread exited.
I, [2020-03-25T05:35:02.636470 #32996]  INFO -- : Disconnecting broker 0
D, [2020-03-25T05:35:02.636522 #32996] DEBUG -- : Closing socket to localhost:9092
I, [2020-03-25T05:35:03.633512 #32996]  INFO -- : [[g] {}:] Leaving group `g`
D, [2020-03-25T05:35:03.633671 #32996] DEBUG -- : [[g] {}:] [leave_group] Opening connection to localhost:9092 with client id c...
D, [2020-03-25T05:35:03.634838 #32996] DEBUG -- : [[g] {}:] [leave_group] Sending leave_group API request 1 to localhost:9092
D, [2020-03-25T05:35:03.635018 #32996] DEBUG -- : [[g] {}:] [leave_group] Waiting for response 1 from localhost:9092
D, [2020-03-25T05:35:03.636950 #32996] DEBUG -- : [[g] {}:] [leave_group] Received response 1 from localhost:9092

After

-- snip --
D, [2020-03-25T05:36:31.482775 #33053] DEBUG -- : [[g] {test: 0, 2, 1}:] Handling fetcher command: stop
I, [2020-03-25T05:36:31.482809 #33053]  INFO -- : [g] {test: 0, 2, 1}: Fetcher thread exited.
I, [2020-03-25T05:36:32.480502 #33053]  INFO -- : [[g] {}:] Leaving group `g`
D, [2020-03-25T05:36:32.480612 #33053] DEBUG -- : [[g] {}:] [leave_group] Sending leave_group API request 6 to localhost:9092
D, [2020-03-25T05:36:32.480760 #33053] DEBUG -- : [[g] {}:] [leave_group] Waiting for response 6 from localhost:9092
D, [2020-03-25T05:36:32.482325 #33053] DEBUG -- : [[g] {}:] [leave_group] Received response 6 from localhost:9092
I, [2020-03-25T05:36:32.482380 #33053]  INFO -- : [[g] {}:] Disconnecting broker 0
D, [2020-03-25T05:36:32.482403 #33053] DEBUG -- : [[g] {}:] Closing socket to localhost:9092

As you can see above, the message "Opening connection to localhost:9092" disappeared in the log after the change.

[dasch]

Thanks!