Skip to content

KAFKA-7211: MM should handle TimeoutException in commitSync #5492

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
huxihx wants to merge 5 commits into from
Closed

KAFKA-7211: MM should handle TimeoutException in commitSync #5492

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0defde1
KAFKA-7211: MM should handle TimeoutException in commitSync
huxi-2b Aug 13, 2018
0228e18
Fix incorrect warning messages on tracking failed commits
huxi-2b Aug 30, 2018
b10c6ac
address lindong28's comments
huxi-2b Aug 30, 2018
b0c33b7
address lindong's comments
huxi-2b Sep 3, 2018
e02bf6e
Fixed the incorrect comment for MM
huxi-2b Sep 4, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion core/src/main/scala/kafka/tools/MirrorMaker.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ import org.apache.kafka.common.record.RecordBatch

import scala.collection.JavaConverters._
import scala.collection.mutable.HashMap
import scala.concurrent.TimeoutException
import scala.util.{Failure, Success, Try}
import scala.util.control.ControlThrowable

/**
Expand Down Expand Up @@ -280,6 +282,10 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
consumerWrapper.commit()
throw e

case _: TimeoutException =>
warn("Failed to commit offsets because the offset commit request processing can not be completed in time. " +

@lindong28 lindong28 Aug 23, 2018

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Related to the concern of data loss, would it be useful to also keep track and log the time of last successful commit in the warning message, so that SRE can gauge how much time worth of data has been duplicated?

s"If you see this regularly, it could indicate that you need to increase the consumer's ${ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG}")

case _: CommitFailedException =>
warn("Failed to commit offsets because the consumer group has rebalanced and assigned partitions to " +
"another instance. If you see this regularly, it could indicate that you need to either increase " +
Expand Down Expand Up @@ -473,7 +479,14 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
}

def commit() {
consumer.commitSync(offsets.map { case (tp, offset) => (tp, new OffsetAndMetadata(offset, ""))}.asJava)
val existingTopics: Set[String] = Try(consumer.listTopics) match {

@omkreddy omkreddy Aug 17, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we do this in case of Timeout error and clear offsets map? Checking for every commit maybe costly.

@huxihx huxihx Aug 20, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@omkreddy Thanks for the response. Currently, commitSync will retry indefinitely when committing offsets for a nonexistent topic, and user could do nothing to get rid of this. That's why I always check the existence of topics before commitSync is called. Does it make any sense?

@hachikuji hachikuji Aug 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With KIP-266, even commitSync will eventually timeout (see the use of default.api.timeout.ms). I prefer @omkreddy's suggestion since listing all the topics in the cluster can be expensive.

@huxihx huxihx Aug 21, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With the trunk code, commitSync indefinitely retried instead of throwing TimeoutException when the topic was deleted. Here were the steps I reproduced:

  1. Create a test topic named test
  2. Write a simple consumer to commit offsets for this topic every 3 seconds using commitSync
  3. Delete this topic
    Later, the consumer complained "Offset commit failed on partition test-0 at offset...." forever. Did I miss anything?

@omkreddy omkreddy Aug 21, 2018 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should get Timeout error after default.api.timeout.ms (default 60secs).
verified with trunk code and its working as expected.

case Success(allTopics) => allTopics.keySet.asInstanceOf[Set[String]]
case Failure(_) => Set.empty
}
if (existingTopics.nonEmpty)
consumer.commitSync(offsets.filterKeys(tp => existingTopics.contains(tp.topic)).map { case (tp, offset) => (tp, new OffsetAndMetadata(offset, "")) }.asJava)

@omkreddy omkreddy Aug 17, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: we can use OffsetAndMetadata(long offset) constructor

@hachikuji hachikuji Aug 20, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also, split to multiple lines. Typical line width is 120 characters.

else
consumer.commitSync(offsets.map { case (tp, offset) => (tp, new OffsetAndMetadata(offset, "")) }.asJava)

@omkreddy omkreddy Aug 17, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: we can use OffsetAndMetadata(long offset) construtor

offsets.clear()
}
}
Expand Down