SpringKafka1.2.2优雅关机

Question

我用的是spring kafka 1.2.2。释放目前，我已经为没有BackOffPolicy和AlwaysRetryPolicy的容器配置了重试模板。确认模式为手动或立即。

当一个SIGTERM出现时，我会让当前消息被处理，当@KafkaListener再次被调用时，我会在容器上抛出RuntimeException，该容器会无限期重试并持续抛出异常。一段时间后，SIGKILL被发出，容器被停止（我认为有更好的方法）。但在这个过程中，消费者会在重启后检索重试的消息，但不会调用KafkaListener进行提交。请参见堆栈跟踪下方的偏移=13

堆栈跟踪：

[20 May 2018 22:37:20] [ INFO] [] [ConsumerCoordinator  onJoinComplete]:[262 ] - Setting newly assigned partitions [messages-0] for group listener
[20 May 2018 22:37:20] [DEBUG] [] [KafkaMessageListenerContainer$ListenerConsumer$2 onPartitionsAssigned]:[513 ] - Committing on assignment: {messages-0=OffsetAndMetadata{offset=13, metadata=''}}
[20 May 2018 22:37:20] [ INFO] [] [AbstractMessageListenerContainer$2 onPartitionsAssigned]:[278 ] - partitions assigned:[messages-0]
[20 May 2018 22:37:20] [DEBUG] [] [KafkaMessageListenerContainer$ListenerConsumer        run]:[632 ] - Received: 0 records
[20 May 2018 22:37:20] [DEBUG] [] [KafkaMessageListenerContainer$ListenerConsumer        run]:[632 ] - Received: 1 records
[20 May 2018 22:37:20] [TRACE] [] [KafkaMessageListenerContainer$ListenerConsumer doInvokeWithRecords]:[931 ] - Processing ConsumerRecord(topic = messages, partition = 0, offset = 13, CreateTime = 1526855737241, serialized key size = 31, serialized value size = 2032, headers = RecordHeaders(headers = [], isReadOnly = false), key = "some key", value = "some random data")
[20 May 2018 22:37:20] [DEBUG] [] [KafkaMessageListenerContainer$ListenerConsumer ackImmediate]:[749 ] - Committing: {messages-0=OffsetAndMetadata{offset=14, metadata=''}

当我看到SIGTERM被发出时，有没有更好的方法来阻止容器，这样@KafkaListener就不会被消息调用了。我知道后来的版本(

任何帮助都非常感谢。

Answer 1

endpoint注册表从1.0开始就存在了。建议1. x用户升级到最新的1.3. x；请参阅此处。

当从侦听器停止容器时，最好在新线程上停止，否则停止将被延迟。

参见2.1。x Containers ToppingErrorHandler了解如何做到这一点。但是，当然，停止后不需要抛出异常。

但是有1个。x、 您将需要丢弃已获取的任何后续消息。