GitHub user shanthoosh opened a pull request:
https://github.com/apache/samza/pull/884
SAMZA-2073: Do not commit the task offsets when shutting down the
SamzaContainer.
SAMZA-1489 added support for committing the offsets of all the running
tasks in a SamzaContainer. Other components in samza such as a
CheckpointListener's aren't developed to account for possibility of commit
after the consumers are stopped.
This unnecessarily results in a unclean shutdown of a samza standalone
processor during the rebalancing phase. Here're the sample logs:
```
apache.samza.container.SamzaContainer@129d533c to shutdown.
2019/01/15 02:38:09.738 INFO [SamzaContainer] [Samza StreamProcessor
Container Thread-0] [hello-brooklin-task] [] Shutting down SamzaContainer.
2019/01/15 02:38:09.738 INFO [SamzaContainer] [Samza StreamProcessor
Container Thread-0] [hello-brooklin-task] [] Shutting down consumer multiplexer.
2019/01/15 02:38:09.741 INFO [KafkaProducer]
[hello-brooklin-task-i001-auditor] [hello-brooklin-task] [] [Producer
clientId=hello-brooklin-task-i001-auditor, transactionalId=nullClosing the
Kafka producer with timeoutMillis = 9223370489334886066 ms.
2019/01/15 02:38:09.748 INFO [SamzaContainer] [Samza StreamProcessor
Container Thread-0] [hello-brooklin-task] [] Shutting down task instance stream
tasks.
2019/01/15 02:38:09.748 INFO [SamzaContainer] [Samza StreamProcessor
Container Thread-0] [hello-brooklin-task] [] Shutting down timer executor
2019/01/15 02:38:09.749 INFO [SamzaContainer] [Samza StreamProcessor
Container Thread-0] [hello-brooklin-task] [] Committing offsets for all task
instances
2019/01/15 02:38:09.753 ERROR [SamzaContainer] [Samza StreamProcessor
Container Thread-0] [hello-brooklin-task] [] Caught exception/error while
shutting down container.
java.lang.IllegalStateException: This consumer has already been closed.
at
org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:1735)
at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1214)
at
com.linkedin.kafka.liclients.consumer.LiKafkaConsumerImpl.commitOffsets(LiKafkaConsumerImpl.java:311)
at
com.linkedin.kafka.liclients.consumer.LiKafkaConsumerImpl.commitSync(LiKafkaConsumerImpl.java:282)
at
com.linkedin.brooklin.client.BaseConsumerImpl.commit(BaseConsumerImpl.java:136)
```
Since the final commit is not critical, it will be better to not do it as a
part of the SamzaContainer shutdown sequence.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/shanthoosh/samza
remove_task_commit_during_shutdown
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/samza/pull/884.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #884
----
commit 5cb3faf01e9a1bfd4f81fb0d927dc774c8d16dd4
Author: Shanthoosh Venkataraman <spvenkat@...>
Date: 2019-01-16T02:26:05Z
Do not commit the task offsets when the task instances are shutting down.
----
---
