message: Handle rdkafka rebalance in progress errors
diff: @@ -17,6 +17,7 @@ export interface IKafkaConsumerOptions extends Partial<IKafkaBaseOptions> {
  consumeTimeout: number;
  consumeLoopTimeoutDelay: number;
  optimizedRebalance: boolean;
  + commitRetryDelay: number;
  additionalOptions?: kafkaTypes.ConsumerGlobalConfig;
  }
  @@ -46,6 +47,7 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  consumeTimeout: 1000,
  consumeLoopTimeoutDelay: 100,
  optimizedRebalance: false,
  + commitRetryDelay: 1000,
  ...options,
  };
  }
  @@ -110,15 +112,21 @@ export class RdkafkaConsumer extends RdkafkaBase implements IConsumer {
  // 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_ILLEGAL_GENERATION;
  + this.consumerOptions.optimizedRebalance &&
  + (err.code === kafka.CODES.ERRORS.ERR_REBALANCE_IN_PROGRESS ||
  + err.code === kafka.CODES.ERRORS.ERR_ILLEGAL_GENERATION);
  }
  for (const offset of offsets) {
  const deferredCommit = this.pendingCommits.get(offset.partition);
  if (deferredCommit) {
  - if (shouldRetryCommit && this.assignedPartitions.has(offset.partition)) {
  + if (shouldRetryCommit) {
  + setTimeout(() => {
  + if (this.assignedPartitions.has(offset.partition)) {
  // we still own this partition. checkpoint again
  - this.consumer.commit(offset);
  + this.consumer?.commit(offset);
  + }
  + }, this.consumerOptions.commitRetryDelay);
  continue;
  }
  