KAFKA-2334 Guard against non-monotonic offsets in the client by mumrah · Pull Request #5991 · apache/kafka

mumrah · 2018-12-03T18:11:00Z

After a recent leader election, the leaders high-water mark might lag behind the offset at the beginning of the new epoch (as well as the previous leader's HW). This can lead to offsets going backwards from a client perspective, which is confusing and leads to strange behavior in some clients.

This change causes Partition#fetchOffsetForTimestamp to throw an exception to indicate the offsets are not yet available from the leader. For new clients, a new OFFSET_NOT_AVAILABLE error is added. For existing clients, a LEADER_NOT_AVAILABLE is thrown.

This is an implementation of KIP-207

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

This reverts commit 3e75fdf.

hachikuji

Thanks for the patch. Left a few comments.

hachikuji · 2018-12-04T18:09:20Z

+    // Only actually check the HW if this is a client request and _not_ while unclean leader
+    // election is enabled
+    val shouldCheckHW = {
+      !logManager.currentDefaultConfig.uncleanLeaderElectionEnable &&


I think we'd want to do this validation even if unclean leader election is enabled. The point that the KIP was making is that high watermark monotonicity can be violated if unclean leader election is enabled. But we'd still want to minimize such violations to cases when an unclean leader election has actually happened.

Thanks for the clarification, I misunderstood the KIP. I'll take this check out.

hachikuji · 2018-12-04T18:10:20Z

                              currentLeaderEpoch: Optional[Integer],
-                              fetchOnlyFromLeader: Boolean): Option[TimestampAndOffset] = inReadLock(leaderIsrUpdateLock) {
+                              fetchOnlyFromLeader: Boolean,
+                              isFromClient: Boolean): Option[TimestampAndOffset] = inReadLock(leaderIsrUpdateLock) {


I think we may be able to drop this argument. When the isolation level is read_uncommitted, then we are limited by the high watermark. For followers, isolation level will be None.

hachikuji · 2018-12-04T18:10:30Z

+        leaderEpochStartOffsetOpt.isDefined
+    }
+
+    if(shouldCheckHW) {


nit: typically we prefer a space after if

mumrah · 2018-12-06T21:28:28Z

retest this please

hachikuji

Thanks for the updates. Left a few more comments.

cmccabe · 2018-12-07T22:39:21Z

            ISOLATION_LEVEL,
            TOPICS_V4);

+    // V5 bump to include new possible error code


Let's be specific and say that the new error code is OFFSET_NOT_AVAILABLE

cmccabe · 2018-12-07T22:43:56Z

      case None => localReplica.logEndOffset.messageOffset
    }

+    // Only actually check the HW if this is a client request (isolation level is null)


Should be, if isolationLevel is defined, this is a client request, right?

hachikuji

Left one minor comment. Other than that, looks good.

hachikuji · 2018-12-08T00:24:20Z

+        // Fail with LEADER_NOT_AVAILABLE
+        time.sleep(retryBackoffMs);
+        client.prepareResponse(listOffsetRequestMatcher(ListOffsetRequest.LATEST_TIMESTAMP),
+                listOffsetResponse(Errors.OFFSET_NOT_AVAILABLE, 1L, 5L), false);


Should be LEADER_NOT_AVAILABLE?

hachikuji · 2018-12-08T00:29:29Z

By the way, I think the new usage of LEADER_NOT_AVAILABLE works with the error handling in old versions of the java client. Might be a good idea to make sure this won't cause problems for any of the other popular clients. Such as librdkafka. cc @edenhill

dhruvilshah3 · 2018-12-08T09:13:55Z

+    if (isolationLevel.isDefined && leaderEpochStartOffsetOpt.isDefined) {
+      // Wait until the HW has caught up with the start offset from this epoch
+      if (leaderEpochStartOffsetOpt.get > localReplica.highWatermark.messageOffset) {
+        throw Errors.OFFSET_NOT_AVAILABLE.exception(s"Failed to fetch offsets for " +


Does this exception apply for all cases or only when the offset for latest timestamp is requested?

That's a good question. Presumably it is possible for a timestamp query to find an offset between the log end offset and the high watermark as well. I guess the point is that we cannot trust the high watermark upper bound until we can ensure that it cannot have gone backwards. Maybe the only offset you can safely query without this validation is the earliest?

I think refusing to return any offset until the HW has caught up is a reasonable solution.

Hmm seems like this behavior was changed then in 6a56493?

Indeed! After talking through the various cases with @hachikuji, we decided to only enforce the check on latest timestamp offset requests.

Thanks! Would be good to document the change in the KIP, given the following is now out-dated:

The KIP-207 behavior applies to all ListOffsetsRequests, whether they are for the latest offset, the earliest offset, or a time-based offset

edenhill · 2018-12-08T16:08:00Z

@hachikuji
In librdkafka LEADER_NOT_AVAILABLE will trigger a metadata-refresh but propagate the error to the application rather than retry,

NOT_LEADER_FOR_PARTITION on the other hand will trigger metadata-refresh and retry.

So from my point of view the latter would be preferred, but I don't think this matters much in practice.

ijuma · 2018-12-08T17:33:55Z

@edenhill what's the reason for the distinction in librdkafka?

edenhill · 2018-12-09T09:08:11Z

@ijuma No reason other than historical, i.e., old code. We have a backlog item to update to latest protocol request versions and update error handling accordingly.

mumrah · 2018-12-10T14:28:53Z

@edenhill what about new unknown error codes (like the one introduce here)? Does librdkafka have some fallback logic for unknown error codes?

hachikuji

Thanks, I think we're almost there. Just one more comment.

hachikuji · 2018-12-11T00:52:09Z

+      // Throw if the HW has not caught up with the start offset from this epoch
+      leaderEpochStartOffsetOpt
+        .filter(leo => leo > localReplica.highWatermark.messageOffset)
+        .map(leo => Errors.OFFSET_NOT_AVAILABLE.exception(s"Failed to fetch offsets for " +


Could we change this map to foreach and throw from here?

If we are only checking this for the latest case, I guess we could move this into the if block below. However, I am considering whether we should check this for general timestamp queries. The problem is that we may be excluding some committed data from a query if we filter only based on the local high watermark. This might cause a timestamp which was visible to disappear temporarily after a leader change.

So we could change the code slightly below. If fetchOffsetByTimestamp returns an offset that is larger than lastFetchableOffset and leaderEpochStartOffsetOpt is defined, then instead of returning None, we could raise OFFSET_NOT_AVAILABLE. Would that make sense?

I think that makes sense, +1

Ok, I incorporated this logic into the latest commit. I also reorganized the code to make it easier to understand (hopefully)

hachikuji

Thanks, the updated logic looks good. I left a few small stylistic suggestions to consider.

hachikuji · 2018-12-12T00:53:57Z

-        timestamp == ListOffsetRequest.EARLIEST_TIMESTAMP || timestampOffset.offset < lastFetchableOffset
+    val invalidEpochStartOffset: Option[Long] = leaderEpochStartOffsetOpt
+      .filter(_ => isolationLevel.isDefined)
+      .filter(leo => leo > localReplica.highWatermark.messageOffset)


nit: perhaps we could combine these two checks into one filter?

hachikuji · 2018-12-12T01:06:21Z

+    }

-      fetchedOffset.filter(allowed)
+    val epochLogString: String = Some(currentLeaderEpoch.orElse(null))


This feels a little like overkill. Using the toString of Option is probably reasonable.

As an aside, Some(null) should generally be avoided. When working with Option, the expectation is that the value inside it is not null.

Yea, the trouble here was dealing with a Java Optional (it doesn't like Scala lambdas apparently). I'll think of something better

There's a Scala/Java 8 integration library that helps with this. We should consider adding they dependency.

hachikuji · 2018-12-12T14:33:02Z


 import scala.collection.JavaConverters._
-import scala.collection.Map
+import scala.collection.{JavaConverters, Map}


nit: seems we didn't need this?

hachikuji · 2018-12-12T14:34:52Z

+    timestamp match {
+      case ListOffsetRequest.LATEST_TIMESTAMP =>
+        // Special case, if asking for "latest" and HW is lagging behind LEO, throw an error
+        if(invalidEpochStartOffset.isDefined) {


nit: use a match?

hachikuji · 2018-12-12T14:36:40Z

+      case ListOffsetRequest.EARLIEST_TIMESTAMP =>
+        getOffsetByTimestamp
+      case _ =>
+        getOffsetByTimestamp match {


nit: seems we could do a flatMap since we aren't doing anything in the None case.

hachikuji · 2018-12-12T14:45:16Z

-    } else {
-      def allowed(timestampOffset: TimestampAndOffset): Boolean =
-        timestamp == ListOffsetRequest.EARLIEST_TIMESTAMP || timestampOffset.offset < lastFetchableOffset
+    val invalidEpochStartOffset: Option[Long] = leaderEpochStartOffsetOpt


I've been trying to come up with an alternative name. Basically we're in a situation where the local high watermark is not trusted and this is an upper bound on the true high watermark. Maybe maxTrueHighWatermark?

hachikuji · 2018-12-12T19:06:26Z

+          } else {
+            // Either throw or return None depending on epochStartOffset
+            if (maxTrueHighWatermark.isDefined) {
+              throw Errors.OFFSET_NOT_AVAILABLE.exception(s"Failed to fetch offsets for " +


Maybe this message should say something about maxTrueHighWatermark?

hachikuji · 2018-12-12T19:07:51Z

+            Some(timestampAndOffset)
+          } else {
+            // Either throw or return None depending on epochStartOffset
+            if (maxTrueHighWatermark.isDefined) {


I think we could technically do this as a map, but I'll leave it up to you if that seems obscure.

hachikuji · 2018-12-12T19:08:53Z

+      case ListOffsetRequest.EARLIEST_TIMESTAMP =>
+        getOffsetByTimestamp
+      case _ =>
+        getOffsetByTimestamp.flatMap(timestampAndOffset =>


nit: this is usually preferred

getOffsetByTimestamp.flatMap { timestampAndOffset => }

hachikuji · 2018-12-12T19:11:07Z

+    // Or if we request the earliest timestamp, we skip the check
+    offsetAndTimestamp = partition.fetchOffsetForTimestamp(
+      timestamp = ListOffsetRequest.EARLIEST_TIMESTAMP,
+      isolationLevel = None,


We should provide an isolation level for this case?

hachikuji · 2018-12-12T19:12:43Z

+      // Try to get offsets as a client
+      partition.fetchOffsetForTimestamp(
+        timestamp = ListOffsetRequest.LATEST_TIMESTAMP,
+        isolationLevel = Some(IsolationLevel.READ_COMMITTED),


Doesn't make a difference for this test case, but maybe we can use READ_UNCOMMITTED so it does not cause any confusion about the expectation in regard to the last stable offset.

hachikuji · 2018-12-12T19:13:30Z

+      fetchOnlyFromLeader = true
+    )
+    assertTrue(offsetAndTimestamp.isDefined)
+    assertEquals(offsetAndTimestamp.get.offset, 5)


nit: the expected value should be the first argument. There are a few of these.

hachikuji · 2018-12-12T19:17:23Z

+
+    try {
+      // Try to get offsets as a client
+      partition.fetchOffsetForTimestamp(


nit: there's a bit of boilerplate here that we could probably factor out into an inner def. Like maybe

def fetchOffsetForTimestamp(timestamp, isolation): Either[OffsetAndTimestamp, Errors]

Or sth like that.

hachikuji · 2018-12-13T23:01:14Z

+    // Only consider throwing an error if we get a client request (isolationLevel is defined) and the start offset
+    // is lagging behind the high watermark
+    val maybeOffsetsError: Option[ApiException] = leaderEpochStartOffsetOpt
+      .filter(epochLSO => isolationLevel.isDefined && epochLSO > localReplica.highWatermark.messageOffset)


nit: LSO annoyingly can be mean either "last stable offset" or "log start offset". How about just startOffset instead?

hachikuji

LGTM. Thanks for the patch!

…#5991) After a recent leader election, the leaders high-water mark might lag behind the offset at the beginning of the new epoch (as well as the previous leader's HW). This can lead to offsets going backwards from a client perspective, which is confusing and leads to strange behavior in some clients. This change causes Partition#fetchOffsetForTimestamp to throw an exception to indicate the offsets are not yet available from the leader. For new clients, a new OFFSET_NOT_AVAILABLE error is added. For existing clients, a LEADER_NOT_AVAILABLE is thrown. This is an implementation of [KIP-207](https://cwiki.apache.org/confluence/display/KAFKA/KIP-207%3A+Offsets+returned+by+ListOffsetsResponse+should+be+monotonically+increasing+even+during+a+partition+leader+change). Reviewers: Colin Patrick McCabe <colin@cmccabe.xyz>, Dhruvil Shah <dhruvil@confluent.io>, Jason Gustafson <jason@confluent.io>

mumrah added 7 commits November 29, 2018 16:33

Add guard against returning offsets when HW has not caught up

c08b21e

Don't do the HW check when in unclean leader election mode

4f1722d

Unit tests passing

2767f70

Make broker and client behavior the same

3e75fdf

Merge remote-tracking branch 'origin/trunk' into KAFKA-2334

eb068a0

Merge remote-tracking branch 'apache/trunk' into KAFKA-2334

3d2f66a

Revert "Make broker and client behavior the same"

70d43d7

This reverts commit 3e75fdf.

hachikuji self-assigned this Dec 3, 2018

Add ASL to new class

1a0a8c5

hachikuji reviewed Dec 4, 2018

View reviewed changes

mumrah added 3 commits December 5, 2018 09:33

Feedback from PR

b76c543

New protocol version (KAFKA_2_2_IV1)

048b1f7

Merge remote-tracking branch 'apache/trunk' into KAFKA-2334

ae075e7

high-water mark -> high watermark

8f0e8f4

hachikuji reviewed Dec 6, 2018

View reviewed changes

mumrah added 4 commits December 6, 2018 17:32

Reverse the logic for the error case (oops)

c7336cb

Add unit test for offset errors

1b42863

Fix scala compile error

ce40782

Fix another scala compile error

ea49519

cmccabe reviewed Dec 7, 2018

View reviewed changes

Comment thread core/src/main/scala/kafka/api/ApiVersion.scala

cmccabe reviewed Dec 7, 2018

View reviewed changes

hachikuji reviewed Dec 8, 2018

View reviewed changes

dhruvilshah3 reviewed Dec 8, 2018

View reviewed changes

mumrah added 2 commits December 10, 2018 09:29

Fix some comments

47cf1c7

Only do the check for "latest" offset requests

6a56493

hachikuji reviewed Dec 11, 2018

View reviewed changes

Throw the new error instead of return None in one case, code reorg

31de57b

hachikuji reviewed Dec 12, 2018

View reviewed changes

Style feedback from PR

ad61063

hachikuji reviewed Dec 12, 2018

View reviewed changes

mumrah added 2 commits December 13, 2018 11:45

More unit tests, cover an additional error case

ca36eba

Clean up some comments

53dd98b

hachikuji reviewed Dec 13, 2018

View reviewed changes

mumrah added 2 commits December 13, 2018 23:05

Remove LSO reference

9e7a667

Another LSO reference

54d8dd5

hachikuji approved these changes Dec 14, 2018

View reviewed changes

hachikuji merged commit 1522929 into apache:trunk Dec 14, 2018

Uh oh!

Conversation

mumrah commented Dec 3, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

mumrah commented Dec 6, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cmccabe Dec 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji commented Dec 8, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

edenhill commented Dec 8, 2018

Uh oh!

ijuma commented Dec 8, 2018

Uh oh!

edenhill commented Dec 9, 2018

Uh oh!

mumrah commented Dec 10, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

hachikuji Dec 11, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mumrah commented Dec 3, 2018 •

edited

Loading

cmccabe Dec 7, 2018 •

edited

Loading

hachikuji Dec 11, 2018 •

edited

Loading