KAFKA-8248; Ensure time updated before sending transactional request #6613

hachikuji · 2019-04-20T07:59:23Z

This patch fixes a bug in the sending of transactional requests. We need to call KafkaClient.send with an updated current time. Failing to do so can result in an IllegalStateExcepton which leaves the producer effectively dead since the in-flight correlation id has been set, but no request has been sent. To avoid the same problem in the future, we update the in flight correlationId only after sending the request.

Committer Checklist (excluded from commit message)

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

mjsax · 2019-04-20T18:27:32Z

@hachikuji Test failures seem to be related.

hachikuji · 2019-04-20T18:58:03Z

@mjsax Thanks. Fixed now. There was some sloppiness in ReplicaFetcherMockBlockingSend which was caught by the changes in MockClient.

apurvam · 2019-04-22T18:20:56Z

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

                    log.debug("Sending transactional request {} to node {}", requestBuilder, targetNode);
-                    client.send(clientRequest, now);
+                    client.send(clientRequest, currentTimeMs);


It seems to me that this is the main fix. Do we really want to redefine 'now' in this method as you do on L455? It is used in multiple places in run and I wonder if there will be side effects of using a different value here.

Hmm.. We definitely have to update the time after the blocking operations above, but it's a fair point that the old now continues to be used in run afterwards. We actually have the same problem in maybeWaitForProducerId, which is also a blocking operation. I don't see a way around it except to change the logic in run to treat now as more of a startTimeMs.

You're right that accounting for the blocking operations is key.

Initially, I thought the main issue with recomputing the currentTime in every down stream call is that it will likely affect the validity of a bunch of the test cases in SenderTest. But I looked at the cases they should be OK with such a change.

I think standardizing on treating now as more of a startTimeMs makes sense.

Thanks for reviewing. I ended up changing the signature so that it removes the timing assumption. Literally every call to this method was run(time.milliseconds()), so I got rid of the argument and renamed to runOnce.

I like the refactoring too: we had similar pattern of passing now in Streamthread loop but later I refactored it to advanceNowAndComputeLatency whenever a blocking call is made for the same reason.

mjsax · 2019-04-23T00:43:43Z

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

@@ -330,8 +330,9 @@ void run(long now) {
            }
        }

-        long pollTimeout = sendProducerData(now);
-        client.poll(pollTimeout, now);
+        long currentTimeMs = time.milliseconds();


Should this also be now instead of currentTimeMs ?

I actually favor currentTimeMs, but perhaps it is unnecessarily pedantic.

I personally don't care too much. Was just wondering about keeping the code consistent. If now is the current way, we might want to keep it this way -- or rename it everywhere (but maybe in a separate PR). Just my 2 cents. Feel free to ignore.

mjsax · 2019-04-23T00:46:33Z

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

@@ -452,11 +453,12 @@ private boolean maybeSendTransactionalRequest(long now) {
                if (targetNode != null) {
                    if (nextRequestHandler.isRetry())
                        time.sleep(nextRequestHandler.retryBackoffMs());
+                    long currentTimeMs = time.milliseconds();


Should this also be now instead of currentTimeMs ?

apurvam

I left a minor comment, but LGTM otherwise. thanks!

apurvam · 2019-04-24T04:55:26Z

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

-        client.poll(pollTimeout, now);
+        long currentTimeMs = time.milliseconds();
+        long pollTimeout = sendProducerData(currentTimeMs);
+        client.poll(pollTimeout, currentTimeMs);
    }

    private long sendProducerData(long now) {


(This is a rewrite of the comment I thought I had left with my previous review)

With the changes in this patch, the path for the regular produce diverges from the transactional path: the former takes in a value of now from runOnce and uses it all the way through while the latter recomputes it. This is fine because the regular producer path has no blocking calls. It would be good add a comment explaining this at the top level.

guozhangwang · 2019-05-01T02:10:36Z

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

                    log.debug("Sending transactional request {} to node {}", requestBuilder, targetNode);
-                    client.send(clientRequest, now);
+                    client.send(clientRequest, currentTimeMs);


I like the refactoring too: we had similar pattern of passing now in Streamthread loop but later I refactored it to advanceNowAndComputeLatency whenever a blocking call is made for the same reason.

guozhangwang · 2019-05-01T02:17:49Z

core/src/test/scala/unit/kafka/server/epoch/util/ReplicaFetcherMockBlockingSend.scala

@@ -59,6 +65,8 @@ class ReplicaFetcherMockBlockingSend(offsets: java.util.Map[TopicPartition, Epoc
  }

  override def sendRequest(requestBuilder: Builder[_ <: AbstractRequest]): ClientResponse = {
+    if (!NetworkClientUtils.awaitReady(client, sourceNode, time, 500))


Not clear to me why adding this is necessary?

I made changes to MockClient to make its behavior closer to NetworkClient. One of the improvements is that you now have to call ready before calling send, which is just like NetworkClient.

guozhangwang · 2019-05-01T02:21:03Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

+                    return notThrottled(now);
+
+                case CONNECTING:
+                    if (now < readyDelayedUntilMs)


nit: some functions are inlined while some others are wrapped in a private function like isUnreachable. Maybe just put this one as a function as well?

guozhangwang · 2019-05-01T02:21:45Z

clients/src/test/java/org/apache/kafka/clients/MockClient.java

-    // Nodes which have a delay before ultimately succeeding to connect
-    private final TransientSet<Node> delayedReady;
-
+    private final Map<String, ConnectionState> connections = new HashMap<>();


Just to clarify this refactoring is orthogonal to unit test augment for this bug fix specifically right?

I needed it in order to add a good test case.

guozhangwang · 2019-05-01T21:14:52Z

LGTM! I think we can merge post green builds.

hachikuji · 2019-05-02T16:24:25Z

I filed https://issues.apache.org/jira/browse/INFRA-18318 for the jdk11 failure. I am going to go ahead and merge this.

…6613) This patch fixes a bug in the sending of transactional requests. We need to call `KafkaClient.send` with an updated current time. Failing to do so can result in an `IllegalStateExcepton` which leaves the producer effectively dead since the in-flight correlation id has been set, but no request has been sent. To avoid the same problem in the future, we update the in flight correlationId only after sending the request. Reviewers: Matthias J. Sax <matthias@confluent.io>, Apurva Mehta <apurva@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

…pache#6613) This patch fixes a bug in the sending of transactional requests. We need to call `KafkaClient.send` with an updated current time. Failing to do so can result in an `IllegalStateExcepton` which leaves the producer effectively dead since the in-flight correlation id has been set, but no request has been sent. To avoid the same problem in the future, we update the in flight correlationId only after sending the request. Reviewers: Matthias J. Sax <matthias@confluent.io>, Apurva Mehta <apurva@confluent.io>, Guozhang Wang <wangguoz@gmail.com>

hachikuji mentioned this pull request Apr 20, 2019

MINOR: Fix IllegalStateException in Producer #6607

Closed

3 tasks

hachikuji requested a review from rajinisivaram April 21, 2019 23:27

apurvam reviewed Apr 22, 2019

View reviewed changes

mjsax reviewed Apr 23, 2019

View reviewed changes

apurvam approved these changes Apr 23, 2019

View reviewed changes

apurvam reviewed Apr 24, 2019

View reviewed changes

guozhangwang reviewed May 1, 2019

View reviewed changes

Jason Gustafson added 4 commits May 1, 2019 13:56

KAFKA-8248; Ensure time updated before sending transactional request

fd1e079

Fix ReplicaFetcherMockBlockingSend to properly call ready

2bc9b33

Fix use of now following blocking operations

2f474df

Add isReadyDelayed method to MockClient.ConnectionState

bfb6728

hachikuji force-pushed the KAFKA-8248 branch from 329fab1 to bfb6728 Compare May 1, 2019 20:59

hachikuji merged commit c34330c into apache:trunk May 2, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8248; Ensure time updated before sending transactional request #6613

KAFKA-8248; Ensure time updated before sending transactional request #6613

hachikuji commented Apr 20, 2019

mjsax commented Apr 20, 2019

hachikuji commented Apr 20, 2019

apurvam Apr 22, 2019

hachikuji Apr 22, 2019

apurvam Apr 22, 2019

hachikuji Apr 22, 2019 •

edited

Loading

guozhangwang May 1, 2019

mjsax Apr 23, 2019

hachikuji Apr 23, 2019

mjsax Apr 23, 2019

mjsax Apr 23, 2019

apurvam left a comment

apurvam Apr 24, 2019

guozhangwang May 1, 2019

guozhangwang May 1, 2019

hachikuji May 1, 2019

guozhangwang May 1, 2019

guozhangwang May 1, 2019

guozhangwang May 1, 2019

hachikuji May 1, 2019

guozhangwang May 1, 2019

guozhangwang commented May 1, 2019

hachikuji commented May 2, 2019

KAFKA-8248; Ensure time updated before sending transactional request #6613

KAFKA-8248; Ensure time updated before sending transactional request #6613

Conversation

hachikuji commented Apr 20, 2019

Committer Checklist (excluded from commit message)

mjsax commented Apr 20, 2019

hachikuji commented Apr 20, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Apr 22, 2019 • edited Loading

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

apurvam 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

guozhangwang commented May 1, 2019

hachikuji commented May 2, 2019

hachikuji Apr 22, 2019 •

edited

Loading