KAFKA-5793: Tighten up the semantics of the OutOfOrderSequenceException #3865

apurvam · 2017-09-15T01:05:28Z

Description of the solution can be found here: https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Exactly+Once+-+Solving+the+problem+of+spurious+OutOfOrderSequence+errors

apurvam · 2017-09-15T18:53:51Z

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

@@ -62,7 +62,7 @@ public Builder(byte magic,
                       int timeout,
                       Map<TopicPartition, MemoryRecords> partitionRecords,
                       String transactionalId) {
-            super(ApiKeys.PRODUCE, (short) (magic == RecordBatch.MAGIC_VALUE_V2 ? 3 : 2));
+            super(ApiKeys.PRODUCE, (short) (magic == RecordBatch.MAGIC_VALUE_V2 ? ApiKeys.PRODUCE.latestVersion() : 2));


I think the 3 being hard coded here was a bug. Once the desired version is set, that is always the protocol version used. I think we should always use the latestVersion if we are using v2 of the message format.

Yes, makes sense.

apurvam · 2017-09-16T00:35:15Z

This is ready for review now.

cc @hachikuji

ijuma · 2017-09-19T04:19:54Z

@apurvam, I just merged a refactor from @hachikuji (0cf7708) that moves the protocol definition to the request/response classes so you'll have to rebase this PR. Hopefully it's straightforward. :)

tests for insertion in order.

hachikuji

Mostly minor comments. Haven't looked at tests yet.

hachikuji · 2017-09-19T23:11:52Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -591,7 +593,7 @@ private void failBatch(ProducerBatch batch, ProduceResponse.PartitionResponse re

    private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception, boolean adjustSequenceNumbers) {
        if (transactionManager != null) {
-            if (exception instanceof OutOfOrderSequenceException
+            if ((exception instanceof OutOfOrderSequenceException || exception instanceof UnknownProducerException)
                    && !transactionManager.isTransactional()
                    && transactionManager.hasProducerId(batch.producerId())) {
                log.error("The broker received an out of order sequence number for topic-partition " +


Probably need to update this log message to mention the specific failure.

hachikuji · 2017-09-19T23:30:39Z

clients/src/test/resources/log4j.properties

@@ -19,3 +19,4 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n

 log4j.logger.org.apache.kafka=ERROR
+log4j.logger.org.apache.kafka.clients.producer.internals=TRACE


Inadvertent I assume.

hachikuji · 2017-09-19T23:31:47Z

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java

@@ -62,7 +62,7 @@ public Builder(byte magic,
                       int timeout,
                       Map<TopicPartition, MemoryRecords> partitionRecords,
                       String transactionalId) {
-            super(ApiKeys.PRODUCE, (short) (magic == RecordBatch.MAGIC_VALUE_V2 ? 3 : 2));
+            super(ApiKeys.PRODUCE, (short) (magic == RecordBatch.MAGIC_VALUE_V2 ? ApiKeys.PRODUCE.latestVersion() : 2));


Yes, makes sense.

hachikuji · 2017-09-19T23:33:07Z

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -524,7 +525,15 @@ public ApiException build(String message) {
            public ApiException build(String message) {
                return new AuthenticationFailedException(message);
            }
-        });
+    }),
+    UNKNOWN_PRODUCER(59, "The specified producerId does not exist on the broker. This indicates data loss on the broker " +


This may indicate data loss, but not necessarily, right?

The idea was that if it is raised to the user, then it definitely indicates data loss. However, based on our offline conversation, we will return the OutOfOrderSequenceException in place of this one to indicate data loss. So I will change the wording of this comment.

hachikuji · 2017-09-19T23:41:29Z

clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java

@@ -156,8 +178,11 @@ public ProduceResponse(Struct struct) {
                Errors error = Errors.forCode(partRespStruct.get(ERROR_CODE));
                long offset = partRespStruct.getLong(BASE_OFFSET_KEY_NAME);
                long logAppendTime = partRespStruct.getLong(LOG_APPEND_TIME_KEY_NAME);
+                long logStartOffset = INVALID_OFFSET;
+                if (partRespStruct.hasField(LOG_START_OFFSET_KEY_NAME))


Maybe we could use the fancy new getOrElse method in Struct.

hachikuji · 2017-09-20T00:15:56Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

@@ -486,6 +508,20 @@ synchronized void adjustSequencesDueToFailedBatch(ProducerBatch batch) {
        }
    }

+    private synchronized void startSequencesAtBeginning(TopicPartition topicPartition) {


Discussed offline. There are some edge cases where resetting the sequence number may not be safe. In particular, since retention policies are enforced asynchronously on the brokers, it could be possible to see an UnknownProducer exception on one broker and then a leader failover could effectively restore some of producer state. This case is rare, but maybe the safest way to handle it is to reset the producerId.

The one offshoot of this is that if we reset the producer id here, then the user is certain to hit https://issues.apache.org/jira/browse/KAFKA-5870, which will result in OutOfOrderSequencExceptions for all in flight requests.

I wonder if that inconvenience is worth it. I am on the fence at this point, but leaning slightly toward leaving the code as is until we fix KAFKA-5870.

hachikuji · 2017-09-20T00:17:31Z

clients/src/main/java/org/apache/kafka/common/errors/UnknownProducerException.java

+
+package org.apache.kafka.common.errors;
+
+public class UnknownProducerException extends ApiException {


A javadoc would be helpful.

hachikuji · 2017-09-20T00:18:00Z

clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java

@@ -188,6 +213,8 @@ protected Struct toStruct(short version) {
                        .set(BASE_OFFSET_KEY_NAME, part.baseOffset);
                if (partStruct.hasField(LOG_APPEND_TIME_KEY_NAME))
                    partStruct.set(LOG_APPEND_TIME_KEY_NAME, part.logAppendTime);
+                if (partStruct.hasField(LOG_START_OFFSET_KEY_NAME))
+                    partStruct.set(LOG_START_OFFSET_KEY_NAME, part.logStartOffset);


Use setIfExists?

hachikuji · 2017-09-20T23:51:20Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

+                // We don't want to return the UnknownProducerException to the user because we already have the
+                // OutOfOrderSequenceException to indicate data loss, and there is no point having two of them
+                // indicating the same thing.
+                exception = Errors.OUT_OF_ORDER_SEQUENCE_NUMBER.exception();


I guess another option might be to let UnknownProducerException extend from OutOfOrderSequenceException. The user would still only need to handle the latter.

hachikuji

LGTM pending a few minor changes discussed offline.

apurvam · 2017-09-21T00:36:04Z

Jason and I were discussing the subclassing vs just changing the exception. Technically speaking, UnkownProducer is an instance of OutOfOrderSequence because it indicates that the broker received a sequence/epoch which is not in sequence. The UnknownProducer case is a sub set of the OutOfOrderSequence case in the narrow instance where the first sequence received by the a broker is not 0. This could mean that the state on the broker was lost due to retention, or it could mean data loss.

We disambiguate this in the client using the additional metadata, and then only raise the exception to the user if it is still data loss.

So the user will only have to handle OutOfOrderSequence in all cases, which is the most ideal thing. By retaining the UnknownProducerException and not morphing it to OutOfOrderSequenceException we would have additional data to help us debug, which is also useful.

ijuma · 2017-09-21T00:41:40Z

OK, that's fine. Note that you could retain the additional data by passing UnknownProducer as a cause of OutOfOrderSequenceException. However, since you say that UnknownProducer is really a subtype of OutOfOrderSequenceException, then it seems fine. One last thing: should the name be UnknownProducerId? Or does UnknownProducer represent the issue better?

ijuma · 2017-09-21T00:43:26Z

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

-        });
+    }),
+    UNKNOWN_PRODUCER(59, "The specified producerId does not exist on the broker. This may indicate data loss on the broker " +
+            "and should be investigated.",


This message is more alarming than the exception Javadoc. Either we should make the other one more alarming or this one less. :)

hachikuji · 2017-09-21T00:43:51Z

Yes, the mores specific name UnknownProducerId seems like an improvement.

ijuma

A few more random comments of things I noticed while quickly looking through the PR. Not blockers, can be addressed separately, if necessary.

ijuma · 2017-09-21T00:44:44Z

core/src/main/scala/kafka/cluster/Partition.scala

@@ -518,6 +518,17 @@ class Partition(val topic: String,
    info
  }

+  def logStartOffset : Long = {


Nit: there should be no space before :.

ijuma · 2017-09-21T00:45:28Z

core/src/main/scala/kafka/cluster/Partition.scala

+          return leaderReplica.log.get.logStartOffset
+        case None =>
+          -1
+      }


leaderReplicaIfLocal.map(_.log.get.logStartOffset).getOrElse(-1) is more concise.

ijuma · 2017-09-21T00:46:10Z

core/src/test/scala/unit/kafka/server/EdgeCaseRequestTest.scala

@@ -114,9 +114,9 @@ class EdgeCaseRequestTest extends KafkaServerTestHarness {
    val correlationId = -1
    TestUtils.createTopic(zkUtils, topic, numPartitions = 1, replicationFactor = 1, servers = servers)

-    val version = 2: Short
+    val version = ApiKeys.PRODUCE.latestVersion() : Short


Nit: no need for () and there should be no space before :.

1. Make the error message for the UnknownProducerIdException less drastic. 2. Rename UnknownProducerException to UnknownProducerIdException

ijuma · 2017-09-21T01:12:36Z

Thanks for addressing the comments. Seem fine to me. And feel free to merge whenever you're ready btw. :)

tedyu · 2017-09-21T16:35:28Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java

@@ -591,12 +593,12 @@ private void failBatch(ProducerBatch batch, ProduceResponse.PartitionResponse re

    private void failBatch(ProducerBatch batch, long baseOffset, long logAppendTime, RuntimeException exception, boolean adjustSequenceNumbers) {
        if (transactionManager != null) {
-            if (exception instanceof OutOfOrderSequenceException
+            if ((exception instanceof OutOfOrderSequenceException || exception instanceof UnknownProducerIdException)


Is the || needed ?
UnknownProducerIdException is subclass of OutOfOrderSequenceException

tedyu · 2017-09-21T16:39:15Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+    synchronized long lastAckedOffset(TopicPartition topicPartition) {
+        Long offset = lastAckedOffset.get(topicPartition);
+        if (offset == null)
+            return -1;


nit: using ProduceResponse.INVALID_OFFSET is better

tedyu · 2017-09-21T16:40:08Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+        if (response.baseOffset == ProduceResponse.INVALID_OFFSET)
+            return;
+        long lastOffset = response.baseOffset + batch.recordCount - 1;
+        if (lastOffset > lastAckedOffset(batch.topicPartition))


Should log be added for the else case ?

hachikuji · 2017-09-21T16:43:08Z

@tedyu The PR has been merged. Feel free to submit another PR if you see room for improvement.

tedyu · 2017-09-21T23:20:31Z

clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java

+            return false;
+
+        Errors error = response.error;
+        if (error == Errors.OUT_OF_ORDER_SEQUENCE_NUMBER && !hasUnresolvedSequence(batch.topicPartition) &&


Shouldn't Errors.UNKNOWN_PRODUCER_ID be included here as well ?
Considering the UnknownProducerIdException is a subclass of OutOfOrderSequenceException.

Actually, no. This is the part of the code where we want to disambiguate between UNKNOWN_PRODUCER_ID and OUT_OF_ORDER_SEQUENCE_EXCEPTION.

apurvam force-pushed the KAFKA-5793-tighten-up-out-of-order-sequence-v2 branch from da658a6 to 08c9ca3 Compare September 15, 2017 18:49

apurvam commented Sep 15, 2017

View reviewed changes

apurvam force-pushed the KAFKA-5793-tighten-up-out-of-order-sequence-v2 branch from a42286b to 103de6b Compare September 16, 2017 00:33

Apurva Mehta added 8 commits September 19, 2017 14:38

Initial commit of the client and server code, with minimal tests

9df6270

Fix up logic

351af7b

Add previously forgotten files to git

f3af8bd

Fix build errors after rebase. Fix failing test

3cd20c2

Add client side tests for handling the new UnknownProducer error

50e90a8

Tweaked the insertInOrder logic to be more general. Added more robust

db6bd9c

tests for insertion in order.

Minor code cleanups

76fd0bc

Fix test failures after rebase

1ec1b5b

apurvam force-pushed the KAFKA-5793-tighten-up-out-of-order-sequence-v2 branch from 2a39853 to 1ec1b5b Compare September 19, 2017 22:02

hachikuji reviewed Sep 20, 2017

View reviewed changes

Apurva Mehta added 2 commits September 19, 2017 18:23

Address PR comments

18ec4eb

Add Javadoc for the UnknownProducerException

44dede0

hachikuji reviewed Sep 20, 2017

View reviewed changes

hachikuji approved these changes Sep 21, 2017

View reviewed changes

Address remaining PR comments

669367d

ijuma reviewed Sep 21, 2017

View reviewed changes

Apurva Mehta added 2 commits September 20, 2017 17:49

Address a couple more PR comments:

3aaf6bc

1. Make the error message for the UnknownProducerIdException less drastic. 2. Rename UnknownProducerException to UnknownProducerIdException

Address Ismael's comments

12f2661

asfgit closed this in 9469228 Sep 21, 2017

tedyu reviewed Sep 21, 2017

View reviewed changes

apurvam deleted the KAFKA-5793-tighten-up-out-of-order-sequence-v2 branch September 22, 2017 18:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-5793: Tighten up the semantics of the OutOfOrderSequenceException #3865

KAFKA-5793: Tighten up the semantics of the OutOfOrderSequenceException #3865

apurvam commented Sep 15, 2017 •

edited

Loading

apurvam Sep 15, 2017

hachikuji Sep 19, 2017

apurvam commented Sep 16, 2017

ijuma commented Sep 19, 2017

hachikuji left a comment

hachikuji Sep 19, 2017

hachikuji Sep 19, 2017

hachikuji Sep 19, 2017

hachikuji Sep 19, 2017

apurvam Sep 20, 2017

hachikuji Sep 19, 2017

hachikuji Sep 20, 2017

apurvam Sep 20, 2017

hachikuji Sep 20, 2017

hachikuji Sep 20, 2017

hachikuji Sep 20, 2017

hachikuji left a comment

apurvam commented Sep 21, 2017

ijuma commented Sep 21, 2017

ijuma Sep 21, 2017

hachikuji commented Sep 21, 2017

ijuma left a comment

ijuma Sep 21, 2017

ijuma Sep 21, 2017

ijuma Sep 21, 2017

ijuma commented Sep 21, 2017

tedyu Sep 21, 2017

tedyu Sep 21, 2017

tedyu Sep 21, 2017

hachikuji commented Sep 21, 2017

tedyu Sep 21, 2017

apurvam Sep 22, 2017


		package org.apache.kafka.common.errors;

		public class UnknownProducerException extends ApiException {

KAFKA-5793: Tighten up the semantics of the OutOfOrderSequenceException #3865

KAFKA-5793: Tighten up the semantics of the OutOfOrderSequenceException #3865

Conversation

apurvam commented Sep 15, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apurvam commented Sep 16, 2017

ijuma commented Sep 19, 2017

hachikuji left a 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

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 left a comment

Choose a reason for hiding this comment

apurvam commented Sep 21, 2017

ijuma commented Sep 21, 2017

Choose a reason for hiding this comment

hachikuji commented Sep 21, 2017

ijuma left a 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

ijuma commented Sep 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji commented Sep 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apurvam commented Sep 15, 2017 •

edited

Loading