message: Fix rdkafka consumer hanging when disconnecting
  Calling `consumer.close()` was hanging indefinitely when the
  `optimizedRebalance` config was enabled.
  The issue is the same as
  The `rebalance` method needs to call `consumer.unassign()` when closing.
  It was not doing this because `this.consumer` was being set to
  `undefined` before disconnecting.
diff: @@ -98,6 +98,9 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  this.zooKeeperClient = new this.consumerOptions.zooKeeperClientConstructor(zooKeeperEndpoint);
  }
  + // eslint-disable-next-line prefer-const
  + let consumer: kafkaTypes.KafkaConsumer;
  +
  const options: kafkaTypes.ConsumerGlobalConfig = {
  "metadata.broker.list": this.endpoints.kafka.join(","),
  "socket.keepalive.enable": true,
  @@ -108,12 +111,14 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  "fetch.min.bytes": 1,
  "fetch.max.bytes": 1024 * 1024,
  "offset_commit_cb": true,
  - "rebalance_cb": this.consumerOptions.optimizedRebalance ? this.rebalance.bind(this) : true,
  + "rebalance_cb": this.consumerOptions.optimizedRebalance ?
  + (err: kafkaTypes.LibrdKafkaError, assignments: kafkaTypes.Assignment[]) => this.rebalance(consumer, err, assignments) :
  + true,
  ...this.consumerOptions.additionalOptions,
  ...this.sslOptions,
  };
  - const consumer: kafkaTypes.KafkaConsumer = this.consumer =
  + consumer = this.consumer =
  new this.kafka.KafkaConsumer(options, { "auto.offset.reset": "latest" });
  consumer.setDefaultConsumeTimeout(this.consumerOptions.consumeTimeout);
  @@ -435,8 +440,8 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  * The default node-rdkafka consumer rebalance callback with the addition
  * of continuing from the last seen offset for assignments that have not changed
  */
  - private rebalance(err: kafkaTypes.LibrdKafkaError, assignments: kafkaTypes.Assignment[]) {
  - if (!this.consumer) {
  + private rebalance(consumer: kafkaTypes.KafkaConsumer | undefined, err: kafkaTypes.LibrdKafkaError, assignments: kafkaTypes.Assignment[]) {
  + if (!consumer) {
  return;
  }
  @@ -452,13 +457,13 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  }
  }
  - this.consumer.assign(assignments);
  + consumer.assign(assignments);
  } else if (err.code === this.kafka.CODES.ERRORS.ERR__REVOKE_PARTITIONS) {
  - this.consumer.unassign();
  + consumer.unassign();
  }
  } catch (ex) {
  - if (this.consumer.isConnected()) {
  - this.consumer.emit("rebalance.error", ex);
  + if (consumer.isConnected()) {
  + consumer.emit("rebalance.error", ex);
  }
  }
  }
  