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

Show file tree
Hide file tree
Changes from 3 commits
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
65 changes: 45 additions & 20 deletions 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 @@ -33,12 +33,13 @@ import org.apache.kafka.clients.producer.internals.ErrorLoggingCallback
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord, RecordMetadata}
import org.apache.kafka.common.{KafkaException, TopicPartition}
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, ByteArraySerializer}
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.errors.WakeupException
import org.apache.kafka.common.utils.{Time, Utils}
import org.apache.kafka.common.errors.{TimeoutException, WakeupException}
import org.apache.kafka.common.record.RecordBatch

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

/**
Expand Down Expand Up @@ -69,6 +70,8 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
private var offsetCommitIntervalMs = 0

@omkreddy omkreddy Sep 3, 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.

not related to this PR. Can we update the comments above:
"There are N mirror maker threads each having one KafkaConsumer instance" ?

private var abortOnSendFailure: Boolean = true
@volatile private var exitingOnSendFailure: Boolean = false
private var lastSuccessfulCommitTime = -1L
private val time = Time.SYSTEM

// If a message send failed after retries are exhausted. The offset of the messages will also be removed from
// the unacked offset list to avoid offset commit being stuck on that offset. In this case, the offset of that
Expand Down Expand Up @@ -267,24 +270,45 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
consumers.map(consumer => new ConsumerWrapper(consumer, customRebalanceListener, whitelist))
}

def commitOffsets(consumerWrapper: ConsumerWrapper) {
def commitOffsets(consumerWrapper: ConsumerWrapper): Unit = {
if (!exitingOnSendFailure) {
trace("Committing offsets.")
try {
consumerWrapper.commit()
} catch {
case e: WakeupException =>
// we only call wakeup() once to close the consumer,
// so if we catch it in commit we can safely retry
// and re-throw to break the loop
var retry = 0
var retryNeeded = true
while (retryNeeded) {
trace("Committing offsets.")
try {
consumerWrapper.commit()
throw e
lastSuccessfulCommitTime = time.milliseconds
retryNeeded = false
} catch {
case e: WakeupException =>
// we only call wakeup() once to close the consumer,
// so if we catch it in commit we can safely retry
// and re-throw to break the loop
commitOffsets(consumerWrapper)
throw e

case _: TimeoutException =>
Try(consumerWrapper.consumer.listTopics) match {
case Success(visibleTopics) =>
consumerWrapper.offsets.retain((tp, _) => visibleTopics.containsKey(tp.topic))
case Failure(e) =>
warn("Failed to list all authorized topics after committing offsets timed out: ", e)
}

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 " +
s"the consumer's ${ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG} or reduce the number of records " +
s"handled on each iteration with ${ConsumerConfig.MAX_POLL_RECORDS_CONFIG}")
retry += 1
warn("Failed to commit offsets because the offset commit request processing can not be completed in time. " +
s"If you see this regularly, it could indicate that you need to increase the consumer's ${ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG} " +
s"Last successful offset commit timestamp=${lastSuccessfulCommitTime}, retry count=${retry}")

@lindong28 lindong28 Aug 31, 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.

nits: we can replace ${lastSuccessfulCommitTime} with $lastSuccessfulCommitTime. Same for ${retry}

Thread.sleep(100)

case _: CommitFailedException =>
retryNeeded = false
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 " +
s"the consumer's ${ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG} or reduce the number of records " +
s"handled on each iteration with ${ConsumerConfig.MAX_POLL_RECORDS_CONFIG}")
}
}
} else {
info("Exiting on send failure, skip committing offsets.")
Expand Down Expand Up @@ -422,14 +446,15 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
}

// Visible for testing
private[tools] class ConsumerWrapper(consumer: Consumer[Array[Byte], Array[Byte]],
private[tools] class ConsumerWrapper(private[tools] val consumer: Consumer[Array[Byte], Array[Byte]],
customRebalanceListener: Option[ConsumerRebalanceListener],
whitelistOpt: Option[String]) {
val regex = whitelistOpt.getOrElse(throw new IllegalArgumentException("New consumer only supports whitelist."))
var recordIter: java.util.Iterator[ConsumerRecord[Array[Byte], Array[Byte]]] = null

// We manually maintain the consumed offsets for historical reasons and it could be simplified
private val offsets = new HashMap[TopicPartition, Long]()
// Visible for testing
private[tools] val offsets = new HashMap[TopicPartition, Long]()

def init() {
debug("Initiating consumer")
Expand Down Expand Up @@ -473,7 +498,7 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
}

def commit() {
consumer.commitSync(offsets.map { case (tp, offset) => (tp, new OffsetAndMetadata(offset, ""))}.asJava)
consumer.commitSync(offsets.map { case (tp, offset) => (tp, new OffsetAndMetadata(offset)) }.asJava)
offsets.clear()
}
}
Expand Down
30 changes: 30 additions & 0 deletions core/src/test/scala/integration/kafka/tools/MirrorMakerIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,15 +24,45 @@ import kafka.tools.MirrorMaker.{ConsumerWrapper, MirrorMakerProducer}
import kafka.utils.TestUtils
import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
import org.apache.kafka.clients.producer.{ProducerConfig, ProducerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.errors.TimeoutException
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, ByteArraySerializer}
import org.junit.Test
import org.junit.Assert._

class MirrorMakerIntegrationTest extends KafkaServerTestHarness {

override def generateConfigs: Seq[KafkaConfig] =
TestUtils.createBrokerConfigs(1, zkConnect).map(KafkaConfig.fromProps(_, new Properties()))

@Test(expected = classOf[TimeoutException])
def testCommitOffsetsThrowTimeoutException(): Unit = {
val consumerProps = new Properties
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group")
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
consumerProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "1")
val consumer = new KafkaConsumer(consumerProps, new ByteArrayDeserializer, new ByteArrayDeserializer)
val mirrorMakerConsumer = new ConsumerWrapper(consumer, None, whitelistOpt = Some("any"))
mirrorMakerConsumer.offsets.put(new TopicPartition("test", 0), 0L)
mirrorMakerConsumer.commit()
}

@Test
def testCommitOffsetsRemoveNonExistentTopics(): Unit = {
val consumerProps = new Properties
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group")
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList)
consumerProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "2000")
val consumer = new KafkaConsumer(consumerProps, new ByteArrayDeserializer, new ByteArrayDeserializer)
val mirrorMakerConsumer = new ConsumerWrapper(consumer, None, whitelistOpt = Some("any"))
mirrorMakerConsumer.offsets.put(new TopicPartition("nonexistent-topic1", 0), 0L)
mirrorMakerConsumer.offsets.put(new TopicPartition("nonexistent-topic2", 0), 0L)
MirrorMaker.commitOffsets(mirrorMakerConsumer)
assertTrue("Offsets for non-existent topics should be removed", mirrorMakerConsumer.offsets.isEmpty)
}

@Test
def testCommaSeparatedRegex(): Unit = {
val topic = "new-topic"
Expand Down