KAFKA-5435: Improve producer state loading after failure #3361

hachikuji · 2017-06-17T00:35:00Z

No description provided.

asfgit · 2017-06-17T00:48:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5427/
Test FAILed (JDK 7 and Scala 2.11).

asfgit · 2017-06-17T00:57:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5412/
Test FAILed (JDK 8 and Scala 2.12).

apurvam · 2017-06-17T01:03:03Z

core/src/main/scala/kafka/log/Log.scala

+    // There are two common cases where we can skip loading and write a new snapshot at the log end offset:
+    //
+    // 1. The broker has been upgraded, but the topic is still on the old message format.
+    // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown.


This is true because we should always have a snapshot file on a clean shutdown with the new message format, correct? In other words, are we assuming that if there isn't a snapshot file and the message format is new and we had a clean shutdown, that means this is an upgrade?

Yes, that should be the common case for a direct upgrade to the new message format.

apurvam · 2017-06-17T01:11:02Z

core/src/main/scala/kafka/log/Log.scala

@@ -445,17 +448,20 @@ class Log(@volatile var dir: File,
        producerStateManager.takeSnapshot()
      }
    } else {
-      val currentTimeMs = time.milliseconds
-      producerStateManager.truncateAndReload(logStartOffset, lastOffset, currentTimeMs)
+      val isEmptyBeforeTruncation = producerStateManager.isEmpty && producerStateManager.mapEndOffset >= lastOffset


Will this always be false for lastOffset > 0 because the producer state is actually loaded on the next line?

That's only the case when doing the initial loading. Truncation after initial load would hit this case.

ah. right. I keep forgetting that leader failover can result in truncation at any time.

I'll add a comment about this because I forgot several times as well.

apurvam · 2017-06-17T01:20:28Z

core/src/main/scala/kafka/log/Log.scala

+        updateFirstUnstableOffset()
+      }
+    } else {
+      lock synchronized {


This is a question you can ignore for now since it's for my understanding: but how could the logStartOffset be in the middle of a segment if numToDelete == 0 ?

asfgit · 2017-06-17T01:34:27Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5429/
Test FAILed (JDK 7 and Scala 2.11).

asfgit · 2017-06-17T01:40:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5414/
Test FAILed (JDK 8 and Scala 2.12).

junrao

@hachikuji : Thanks for the patch. LGTM. Just a couple of minor comments.

junrao · 2017-06-17T00:55:19Z

core/src/main/scala/kafka/log/Log.scala

+    // There are two common cases where we can skip loading and write a new snapshot at the log end offset:
+    //
+    // 1. The broker has been upgraded, but the topic is still on the old message format.
+    // 2. The broker has been upgraded, the topic is on the new message format, and we had a clean shutdown.


Perhaps make it clear that both 1 and 2 are when there is no snapshot.

junrao · 2017-06-17T00:56:17Z

core/src/main/scala/kafka/log/Log.scala

+
+    if (producerStateManager.latestSnapshotOffset.isEmpty && (messageFormatVersion < RecordBatch.MAGIC_VALUE_V2 || reloadFromCleanShutdown)) {
+      // To avoid an expensive scan through all of the segments, we take empty snapshots from the start of
+      // the last two segments and the last offset. This avoid the full scan in the case that the log needs


avoid => avoids

asfgit · 2017-06-17T02:48:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5417/
Test PASSed (JDK 8 and Scala 2.12).

hachikuji · 2017-06-17T02:57:19Z

By the way, I ran the transaction system tests here: https://jenkins.confluent.io/view/All/job/system-test-kafka-branch-builder/917/console. I'll run a few more times.

asfgit · 2017-06-17T03:06:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5432/
Test PASSed (JDK 7 and Scala 2.11).

hachikuji · 2017-06-17T03:30:07Z

core/src/main/scala/kafka/log/Log.scala

+      lock synchronized {
+        // The log start offset may now point to the middle of a segment, so we need to truncate the producer
+        // state to ensure that non-retained producers are evicted.
+        producerStateManager.truncateHead(logStartOffset)


@junrao One question I had is whether it is more appropriate to try and handle this case in maybeIncrementLogStartOffset?

logStartOffset is only going to change when some segments are deleted in deleteSegments() or when maybeIncrementLogStartOffset() is called. So, it seems that we don't need to do anything in the else clause since the other two cases are already covered.

Hmm.. Currently I don't have any logic in maybeIncrementLogStartOffset. Maybe this block needs to move there?

Yes, perhaps we can move the following 2 lines to maybeIncrementLogStartOffset() and have deleteSegments() call maybeIncrementLogStartOffset() instead of directly updating logStartOffset.

producerStateManager.truncateHead(logStartOffset) updateFirstUnstableOffset()

junrao · 2017-06-17T04:50:46Z

core/src/main/scala/kafka/log/ProducerStateManager.scala

@@ -551,21 +556,33 @@ class ProducerStateManager(val topicPartition: TopicPartition,
   */
  def oldestSnapshotOffset: Option[Long] = oldestSnapshotFile.map(file => offsetFromFilename(file.getName))

+  private def isProducerRetained(producerIdEntry: ProducerIdEntry, logStartOffset: Long): Boolean = {
+    producerIdEntry.lastOffset >= logStartOffset


So, when logStartOffset advances, we don't update producerIdEntry.currentTxnFirstOffset, which will be used to compute lastStableOffset. Could that lead to the case that lastStableOffset is < logStartOffset?

Yes, we discussed this briefly before. I think we agreed that we would keep the currentTxnFirstOffset preserved (since it is difficult to find the next higher offset for a transaction), but we would fix first unstable offset to return no lower than log start offset. Does that sound correct?

Yes, that sounds good.

@junrao I added some code to do this. See my comment in the commit. I fear we will have to live with the possibility of this allowing the LSO to temporarily diverge between replicas. Hopefully this is rare. In any case, there seems little else we can do about it this late in the game (though I'm open to ideas).

asfgit · 2017-06-17T04:54:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5434/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-06-17T05:14:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5419/
Test PASSed (JDK 8 and Scala 2.12).

junrao · 2017-06-17T05:38:23Z

core/src/main/scala/kafka/log/Log.scala

-        producerStateManager.evictUnretainedProducers(logStartOffset)
-        updateFirstUnstableOffset()
+        val newLogStartOffset = math.max(logStartOffset, segments.firstEntry.getValue.baseOffset)
+        leaderEpochCache.clearAndFlushEarliest(newLogStartOffset)


It seems that both line 1070 and 1071 can be moved into maybeIncrementLogStartOffset(). We can just call maybeIncrementLogStartOffset(segments.firstEntry.getValue.baseOffset) here. Inside maybeIncrementLogStartOffset, if startOffset changes, we do the leaderEpoch and producerState update.

asfgit · 2017-06-17T08:04:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5443/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-06-17T08:28:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5428/
Test PASSed (JDK 8 and Scala 2.12).

junrao · 2017-06-17T15:11:11Z

@hachikuji : Thanks for the updated patch. LGTM. I will let you merge it.

hachikuji · 2017-06-17T17:18:28Z

I ran the transactions system tests last night and we're looking good: http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2017-06-17--001.1497703753--hachikuji--KAFKA-5435-ALT--49205c7/report.txt. I will go ahead and merge this to trunk and 0.11.0.

Author: Jason Gustafson <jason@confluent.io> Reviewers: Apurva Mehta <apurva@confluent.io>, Jun Rao <junrao@gmail.com> Closes #3361 from hachikuji/KAFKA-5435-ALT (cherry picked from commit bcaee7f) Signed-off-by: Jason Gustafson <jason@confluent.io>

KAFKA-5435: Improve producer state loading after failure

5bf7d18

Fix failing LogManager test

06568e8

apurvam reviewed Jun 17, 2017

View reviewed changes

Revert change in MockScheduler and tune maxLogAgeMs in LogManagerTest

461b8ab

junrao reviewed Jun 17, 2017

View reviewed changes

hachikuji commented Jun 17, 2017

View reviewed changes

Improve comments and add test case for empty producer state optimization

8b85921

junrao reviewed Jun 17, 2017

View reviewed changes

Move producer state truncateHead into maybeIncrementLogStartOffset

4b39398

junrao reviewed Jun 17, 2017

View reviewed changes

Jason Gustafson added 3 commits June 16, 2017 23:02

Move leader epoch cache flush into maybeIncrementLogStartOffset

f33d403

Use log start offset as a lower bound for first unstable offset

90c03a2

fix Utils min and max comments

49205c7

asfgit closed this in bcaee7f Jun 17, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-5435: Improve producer state loading after failure #3361

KAFKA-5435: Improve producer state loading after failure #3361

hachikuji commented Jun 17, 2017

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

apurvam Jun 17, 2017

hachikuji Jun 17, 2017

apurvam Jun 17, 2017

hachikuji Jun 17, 2017

apurvam Jun 17, 2017

hachikuji Jun 17, 2017

apurvam Jun 17, 2017

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

junrao left a comment

junrao Jun 17, 2017

junrao Jun 17, 2017

asfgit commented Jun 17, 2017

hachikuji commented Jun 17, 2017

asfgit commented Jun 17, 2017

hachikuji Jun 17, 2017

junrao Jun 17, 2017

hachikuji Jun 17, 2017

junrao Jun 17, 2017

hachikuji Jun 17, 2017

junrao Jun 17, 2017

hachikuji Jun 17, 2017

junrao Jun 17, 2017

hachikuji Jun 17, 2017 •

edited

Loading

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

junrao Jun 17, 2017

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

junrao commented Jun 17, 2017

hachikuji commented Jun 17, 2017

KAFKA-5435: Improve producer state loading after failure #3361

KAFKA-5435: Improve producer state loading after failure #3361

Conversation

hachikuji commented Jun 17, 2017

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

junrao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfgit commented Jun 17, 2017

hachikuji commented Jun 17, 2017

asfgit commented Jun 17, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Jun 17, 2017 • edited Loading

Choose a reason for hiding this comment

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

Choose a reason for hiding this comment

asfgit commented Jun 17, 2017

asfgit commented Jun 17, 2017

junrao commented Jun 17, 2017

hachikuji commented Jun 17, 2017

hachikuji Jun 17, 2017 •

edited

Loading