message: Add support for manual message consuming in rdkafka
diff: @@ -18,6 +18,7 @@ export interface IKafkaConsumerOptions extends Partial<IKafkaBaseOptions> {
  consumeLoopTimeoutDelay: number;
  optimizedRebalance: boolean;
  commitRetryDelay: number;
  + automaticConsume: boolean;
  additionalOptions?: kafkaTypes.ConsumerGlobalConfig;
  }
  @@ -48,6 +49,7 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  consumeLoopTimeoutDelay: 100,
  optimizedRebalance: false,
  commitRetryDelay: 1000,
  + automaticConsume: true,
  ...options,
  };
  }
  @@ -84,9 +86,13 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  this.consumer.on("ready", () => {
  this.consumer.subscribe([this.topic]);
  +
  + if (this.consumerOptions.automaticConsume) {
  + // start the consume loop
  this.consumer.consume();
  + }
  - this.emit("connected");
  + this.emit("connected", this.consumer);
  });
  this.consumer.on("disconnected", () => {
  @@ -107,14 +113,16 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  let shouldRetryCommit = false;
  if (err) {
  - this.emit("error", err);
  -
  // a rebalance occurred while we were committing
  // we can resubmit the commit if we still own the partition
  shouldRetryCommit =
  this.consumerOptions.optimizedRebalance &&
  (err.code === kafka.CODES.ERRORS.ERR_REBALANCE_IN_PROGRESS ||
  err.code === kafka.CODES.ERRORS.ERR_ILLEGAL_GENERATION);
  +
  + if (!shouldRetryCommit) {
  + this.emit("error", err);
  + }
  }
  for (const offset of offsets) {
  