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-5913; Add hasOffset() and hasTimestamp() methods to RecordMetadata
Date Thu, 21 Sep 2017 21:40:00 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 690575ec4 -> ea533f0c5


KAFKA-5913; Add hasOffset() and hasTimestamp() methods to RecordMetadata

These methods help users check for cases in which this metadata was not returned by the broker
(e.g. in the case of acks=0 or a duplicate error when idempotence is enabled).

Author: Apurva Mehta <apurva@confluent.io>

Reviewers: Ismael Juma <ismael@juma.me.uk>, Jason Gustafson <jason@confluent.io>

Closes #3878 from apurvam/KAFKA-5913-add-record-metadata-not-available-exception


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

Branch: refs/heads/trunk
Commit: ea533f0c5e2438f0aef070dbadb023bffb9ce009
Parents: 690575e
Author: Apurva Mehta <apurva@confluent.io>
Authored: Thu Sep 21 14:30:37 2017 -0700
Committer: Jason Gustafson <jason@confluent.io>
Committed: Thu Sep 21 14:39:00 2017 -0700

----------------------------------------------------------------------
 .../kafka/clients/producer/RecordMetadata.java  | 23 +++++++++++++++++++-
 .../clients/producer/RecordMetadataTest.java    |  2 ++
 .../clients/producer/internals/SenderTest.java  |  6 ++++-
 .../kafka/api/ProducerFailureHandlingTest.scala |  6 ++++-
 4 files changed, 34 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/ea533f0c/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
----------------------------------------------------------------------
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java b/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
index 6757a6d..0924244 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/RecordMetadata.java
@@ -18,6 +18,8 @@ package org.apache.kafka.clients.producer;
 
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.DefaultRecord;
+import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.requests.ProduceResponse;
 
 /**
  * The metadata for a record that has been acknowledged by the server
@@ -65,17 +67,36 @@ public final class RecordMetadata {
     }
 
     /**
+     * Indicates whether the record metadata includes the offset.
+     * @return true if the offset is included in the metadata, false otherwise.
+     */
+    public boolean hasOffset() {
+        return this.offset != ProduceResponse.INVALID_OFFSET;
+    }
+
+    /**
      * The offset of the record in the topic/partition.
+     * @return the offset of the record, or -1 if {{@link #hasOffset()}} returns false.
      */
     public long offset() {
         return this.offset;
     }
 
     /**
+     * Indicates whether the record metadata includes the timestamp.
+     * @return true if a valid timestamp exists, false otherwise.
+     */
+    public boolean hasTimestamp() {
+        return this.timestamp != RecordBatch.NO_TIMESTAMP;
+    }
+
+    /**
      * The timestamp of the record in the topic/partition.
+     *
+     * @return the timestamp of the record, or -1 if the {{@link #hasTimestamp()}} returns
false.
      */
     public long timestamp() {
-        return timestamp;
+        return this.timestamp;
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/kafka/blob/ea533f0c/clients/src/test/java/org/apache/kafka/clients/producer/RecordMetadataTest.java
----------------------------------------------------------------------
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/RecordMetadataTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/RecordMetadataTest.java
index a735a61..bc3ffc7 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/RecordMetadataTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/RecordMetadataTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.record.DefaultRecord;
 import org.junit.Test;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 
 public class RecordMetadataTest {
 
@@ -37,6 +38,7 @@ public class RecordMetadataTest {
         assertEquals(tp.topic(), metadata.topic());
         assertEquals(tp.partition(), metadata.partition());
         assertEquals(timestamp, metadata.timestamp());
+        assertFalse(metadata.hasOffset());
         assertEquals(-1L, metadata.offset());
         assertEquals(checksum.longValue(), metadata.checksum());
         assertEquals(keySize, metadata.serializedKeySize());

http://git-wip-us.apache.org/repos/asf/kafka/blob/ea533f0c/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 0995e36..e1ea10a 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
@@ -1154,7 +1154,7 @@ public class SenderTest {
         assertEquals(1000, transactionManager.lastAckedOffset(tp0));
         assertEquals(1, transactionManager.lastAckedSequence(tp0));
 
-        client.respondToRequest(firstClientRequest, produceResponse(tp0, -1, Errors.DUPLICATE_SEQUENCE_NUMBER,
-1));
+        client.respondToRequest(firstClientRequest, produceResponse(tp0, ProduceResponse.INVALID_OFFSET,
Errors.DUPLICATE_SEQUENCE_NUMBER, 0));
 
         sender.run(time.milliseconds()); // receive response 0
 
@@ -1162,6 +1162,10 @@ public class SenderTest {
         assertEquals(1, transactionManager.lastAckedSequence(tp0));
         assertEquals(1000, transactionManager.lastAckedOffset(tp0));
         assertFalse(client.hasInFlightRequests());
+
+        RecordMetadata unknownMetadata = request1.get();
+        assertFalse(unknownMetadata.hasOffset());
+        assertEquals(-1L, unknownMetadata.offset());
     }
 
     @Test

http://git-wip-us.apache.org/repos/asf/kafka/blob/ea533f0c/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
index 49a096a..e1b1c9d 100644
--- a/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
+++ b/core/src/test/scala/integration/kafka/api/ProducerFailureHandlingTest.scala
@@ -92,7 +92,11 @@ class ProducerFailureHandlingTest extends KafkaServerTestHarness {
 
     // send a too-large record
     val record = new ProducerRecord(topic1, null, "key".getBytes, new Array[Byte](serverMessageMaxBytes
+ 1))
-    assertEquals("Returned metadata should have offset -1", producer1.send(record).get.offset,
-1L)
+
+    val recordMetadata = producer1.send(record).get()
+    assertNotNull(recordMetadata)
+    assertFalse(recordMetadata.hasOffset)
+    assertEquals(-1L, recordMetadata.offset)
   }
 
   /**


Mime
View raw message