SOHU-Co/kafka-node

ConsumerGroup Noting to be commited

SuperZhangAo opened this issue · 0 comments

fillConsumerGroup.on('message', async function(message) {
  try {
    await ctx.service.consumerTest.start_fill(JSON.parse(message.value));
    fillConsumerGroup.commit(true, (err, data) => {
      logger.error('commit:', err, data);
    });
  } catch (error) {
    logger.error('ERROR: [GetMessage] ', message, error);
  }
});

const fillOptions = {
  kafkaHost: config.kafkaHost, // connect directly to kafka broker (instantiates a KafkaClient)
  batch: undefined, // put client batch settings if you need them
  ssl: false, // optional (defaults to false) or tls options hash
  id: `consumer-fill-${uuid}`,
  groupId: 'ExampleFillGroup',
  sessionTimeout: 15000,
  autoCommit: false,
  // An array of partition assignment protocols ordered by preference.
  // 'roundrobin' or 'range' string for built ins (see below to pass in custom assignment protocol)
  protocol: [ 'roundrobin' ],
  encoding: 'utf8', // default is utf8, use 'buffer' for binary data

  // Offsets to use for new groups other options could be 'earliest' or 'none' (none will emit an error if no offsets were saved)
  // equivalent to Java client's auto.offset.reset
  fromOffset: 'latest', // default
  commitOffsetsOnFirstJoin: true, // on the very first time this consumer group subscribes to a topic, record the offset returned in fromOffset (latest/earliest)
  // how to recover from OutOfRangeOffset error (where save offset is past server retention) accepts same value as fromOffset
  outOfRangeOffset: 'earliest', // default
  fetchMaxBytes: 16777216,
  // Callback to allow consumers with autoCommit false a chance to commit before a rebalance finishes
  // isAlreadyMember will be false on the first connection, and true on rebalances triggered after that
  onRebalance: (isAlreadyMember, callback) => {
    callback();
  }, // or null
};

Why Noting to be commited?