root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 4 Packages Projects Releases Wiki Activity

KAFKA-2784: swallow exceptions when mirror maker exits. 

Author: Jiangjie Qin <becket.qin@gmail.com>

Reviewers: Gwen Shapira

Closes #478 from becketqin/KAFKA-2784
This commit is contained in:
Jiangjie Qin 2016-03-02 18:22:28 -08:00 committed by Gwen Shapira
parent fe1fd703e8
commit 9cd65c46a7
1 changed files with 4 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
core/src/main/scala/kafka/tools/MirrorMaker.scala
View File

@ -425,14 +425,10 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
info("Flushing producer.") info("Flushing producer.")
producer.flush() producer.flush()
info("Committing consumer offsets.") info("Committing consumer offsets.")
try { CoreUtils.swallow(commitOffsets(mirrorMakerConsumer))
commitOffsets(mirrorMakerConsumer)
} catch {
case e: WakeupException => // just ignore
}
info("Shutting down consumer connectors.") info("Shutting down consumer connectors.")
// we do not need to call consumer.close() since the consumer has already been interrupted CoreUtils.swallow(mirrorMakerConsumer.stop())
mirrorMakerConsumer.cleanup() CoreUtils.swallow(mirrorMakerConsumer.cleanup())
shutdownLatch.countDown() shutdownLatch.countDown()
info("Mirror maker thread stopped") info("Mirror maker thread stopped")
// if it exits accidentally, stop the entire mirror maker // if it exits accidentally, stop the entire mirror maker
@ -572,7 +568,7 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
} }
override def cleanup() { override def cleanup() {
ClientUtils.swallow(consumer.close()) consumer.close()
} }
override def commit() { override def commit() {