kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject kafka git commit: KAFKA-5959; Fix NPE in Sender.canRetry when idempotence is not enabled
Date Fri, 22 Sep 2017 20:09:28 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk d60f011d7 -> 8f90fd653


KAFKA-5959; Fix NPE in Sender.canRetry when idempotence is not enabled

Author: Apurva Mehta <apurva@confluent.io>

Reviewers: tedyu <yuzhihong@gmail.com>, Jason Gustafson <jason@confluent.io>

Closes #3947 from apurvam/KAFKA-5959-npe-in-sender


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/8f90fd65
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/8f90fd65
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/8f90fd65

Branch: refs/heads/trunk
Commit: 8f90fd6530ce2c4f7e2fdfe3541c61d0178289d5
Parents: d60f011
Author: Apurva Mehta <apurva@confluent.io>
Authored: Fri Sep 22 13:07:28 2017 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Fri Sep 22 13:07:59 2017 -0700

----------------------------------------------------------------------
 .../clients/producer/internals/Sender.java      |  3 +-
 .../clients/producer/internals/SenderTest.java  | 30 ++++++++++++++++++++
 2 files changed, 32 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/8f90fd65/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
index d71046a..45a2919 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java
@@ -636,7 +636,8 @@ public class Sender implements Runnable {
      */
     private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response)
{
         return batch.attempts() < this.retries &&
-                ((response.error.exception() instanceof RetriableException) || transactionManager.canRetry(response,
batch));
+                ((response.error.exception() instanceof RetriableException) ||
+                        (transactionManager != null && transactionManager.canRetry(response,
batch)));
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/8f90fd65/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
index e1ea10a..ecf77aa 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
@@ -33,6 +33,7 @@ import org.apache.kafka.common.errors.ClusterAuthorizationException;
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
 import org.apache.kafka.common.errors.RecordTooLargeException;
 import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.TopicAuthorizationException;
 import org.apache.kafka.common.errors.UnsupportedForMessageFormatException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.internals.ClusterResourceListeners;
@@ -500,6 +501,35 @@ public class SenderTest {
         assertSendFailure(ClusterAuthorizationException.class);
     }
 
+    @Test
+    public void testCanRetryWithoutIdempotence() throws Exception {
+        // do a successful retry
+        Future<RecordMetadata> future = accumulator.append(tp0, 0L, "key".getBytes(),
"value".getBytes(), null, null, MAX_BLOCK_TIMEOUT).future;
+        sender.run(time.milliseconds()); // connect
+        sender.run(time.milliseconds()); // send produce request
+        String id = client.requests().peek().destination();
+        Node node = new Node(Integer.parseInt(id), "localhost", 0);
+        assertEquals(1, client.inFlightRequestCount());
+        assertTrue(client.hasInFlightRequests());
+        assertTrue("Client ready status should be true", client.isReady(node, 0L));
+        assertFalse(future.isDone());
+
+        client.respond(new MockClient.RequestMatcher() {
+            @Override
+            public boolean matches(AbstractRequest body) {
+                ProduceRequest request = (ProduceRequest) body;
+                assertFalse(request.isIdempotent());
+                return true;
+            }
+        }, produceResponse(tp0, -1L, Errors.TOPIC_AUTHORIZATION_FAILED, 0));
+        sender.run(time.milliseconds());
+        assertTrue(future.isDone());
+        try {
+            future.get();
+        } catch (Exception e) {
+            assertTrue(e.getCause() instanceof TopicAuthorizationException);
+        }
+    }
 
     @Test
     public void testIdempotenceWithMultipleInflights() throws Exception {


Mime
View raw message