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 2 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
33 changes: 26 additions & 7 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.{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,19 +270,34 @@ object MirrorMaker extends Logging with KafkaMetricsGroup {
consumers.map(consumer => new ConsumerWrapper(consumer, customRebalanceListener, whitelist))
}

def commitOffsets(consumerWrapper: ConsumerWrapper) {
def commitOffsets(consumerWrapper: ConsumerWrapper, retry: Int = Integer.MAX_VALUE): Unit = {

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

It seems that MirrorMaker. commitOffsets() is public and thus it is public API of an existing tool. Maybe we need to have a KIP in order to change this API. For now it seems simpler to just keep the old compatible behavior such that MM will block infinitely if commit can not pass due something other than topic deletion. If there is good use-case for other behavior than we can have a KIP. Does this sound OK?

if (!exitingOnSendFailure) {
trace("Committing offsets.")
try {
consumerWrapper.commit()
lastSuccessfulCommitTime = time.milliseconds
} 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
consumerWrapper.commit()
commitOffsets(consumerWrapper, retry)
throw e

case _: TimeoutException if retry > 0 =>
if (retry == Integer.MAX_VALUE) { // only try to remove offsets for nonexistent topics once

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

What if topic deletion happens after a few retries? It seems safer to to filter topic for every retry. It is probably OK because the time used to filter topic in MM should be much smaller than the time needed for backoff and the time needed to wait for offset commit to pass, right?

Try(consumerWrapper.consumer.listTopics) match {
case Success(visibleTopics) =>
consumerWrapper.offsets.retain((tp, _) => visibleTopics.containsKey(tp.topic))
case _ =>
}
}
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} " +
s"Last successful offset commit timestamp=${lastSuccessfulCommitTime}, retry count=${Integer.MAX_VALUE - retry}")
Thread.sleep(100)
commitOffsets(consumerWrapper, retry - 1)

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 @@ -422,14 +440,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 +492,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