我知道在 Sping-Kafka 我们有以下方法：
无效 registerSeekCallback(ConsumerSeekCallback 回调);
void onPartitionsAssigned(Map assignments, ConsumerSeekCallback 回调);
void onIdleContainer(Map assignments, ConsumerSeekCallback 回调);
但是哪一个和原生的 ConsumerRebalanceListener 方法 onPartitionsRevoked 一样呢？
“此方法将在重新平衡操作开始之前和消费者停止获取数据之后调用。建议在此回调中将偏移量提交给 Kafka 或自定义偏移量存储，以防止重复数据。”
如果我想实现 ConsumerRebalanceListener，如何传递 KafkaConsumer 引用？我只看到 Spring-Kafka 的消费者。
=========更新======
嗨，加里，当我将 RebalanceListener 添加到 ContainerProperties 中时。我可以看到这两种方法都被触发了。但是，我遇到了异常，说“提交无法完成，因为该组已经重新平衡并将分区分配给另一个成员。这意味着后续调用 poll() 之间的时间比配置的 max.poll 长。 interval.ms，这通常意味着轮询循环花费了太多时间处理消息“你知道吗？
===========更新2 ============
public ConcurrentMessageListenerContainer<Integer, String> createContainer(
ContainerProperties containerProps, IKafkaConsumer iKafkaConsumer) {
Map<String, Object> props = consumerProps();
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
**RebalanceListner rebalanceListner = new RebalanceListner(cf.createConsumer());**
CustomKafkaMessageListener ckml = new CustomKafkaMessageListener(iKafkaConsumer, rebalanceListner);
CustomRecordFilter cff = new CustomRecordFilter();
FilteringAcknowledgingMessageListenerAdapter faml = new FilteringAcknowledgingMessageListenerAdapter(ckml, cff, true);
SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
retryPolicy.setMaxAttempts(5);
FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(1500); // 1.5 seconds
RetryTemplate rt = new RetryTemplate();
rt.setBackOffPolicy(backOffPolicy);
rt.setRetryPolicy(retryPolicy);
rt.registerListener(ckml);
RetryingAcknowledgingMessageListenerAdapter rml = new RetryingAcknowledgingMessageListenerAdapter(faml, rt);
containerProps.setConsumerRebalanceListener(rebalanceListner);
containerProps.setMessageListener(rml);
containerProps.setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
containerProps.setErrorHandler(ckml);
containerProps.setAckOnError(false);
ConcurrentMessageListenerContainer<Integer, String> container = new ConcurrentMessageListenerContainer<>(
cf, containerProps);
container.setConcurrency(1);
return container;
}