KAFKA-12297: Make MockProducer return RecordMetadata with values as p…

…er contract

This is a simple change to MockProducer as per request in KAFKA-12297.
MockProducer currently returns a null RecordMetadata on Exception. The fix will make MockProducer return the right value as per specification.

This only impacts clients which use send with a custom callback and try to then use the RecordMetadata inspite of getting an exception. This should mostly impact customer unit and integration tests as the mock end point was never intended for use in a real Kafka cluster.

Author: Akhilesh Dubey <adubey@confluent.io>
Author: Manikumar Reddy <manikumar.reddy@gmail.com>

Reviewers: Manikumar Reddy <manikumar.reddy@gmail.com>

Closes apache#10110 from aadubey/trunk

clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java

@@ -314,7 +314,7 @@ public synchronized Future<RecordMetadata> send(ProducerRecord<K, V> record, Cal
                 0L, 0, 0, Time.SYSTEM);
         long offset = nextOffset(topicPartition);
         Completion completion = new Completion(offset, new RecordMetadata(topicPartition, 0, offset,
-                RecordBatch.NO_TIMESTAMP, 0L, 0, 0), result, callback);
+                RecordBatch.NO_TIMESTAMP, 0L, 0, 0), result, callback, topicPartition);
 
         if (!this.transactionInFlight)
             this.sent.add(record);
@@ -512,15 +512,18 @@ private static class Completion {
         private final RecordMetadata metadata;
         private final ProduceRequestResult result;
         private final Callback callback;
+        private final TopicPartition tp;
 
         public Completion(long offset,
                           RecordMetadata metadata,
                           ProduceRequestResult result,
-                          Callback callback) {
+                          Callback callback,
+                          TopicPartition tp) {
             this.metadata = metadata;
             this.offset = offset;
             this.result = result;
             this.callback = callback;
+            this.tp = tp;
         }
 
         public void complete(RuntimeException e) {
@@ -529,7 +532,7 @@ public void complete(RuntimeException e) {
                 if (e == null)
                     callback.onCompletion(metadata, null);
                 else
-                    callback.onCompletion(null, e);
+                    callback.onCompletion(new RecordMetadata(tp, -1, -1, RecordBatch.NO_TIMESTAMP, -1L, -1, -1), e);
             }
             result.done();
         }

clients/src/test/java/org/apache/kafka/clients/producer/MockProducerTest.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.ProducerFencedException;
+import org.apache.kafka.common.record.RecordBatch;
 import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.apache.kafka.test.MockSerializer;
@@ -42,6 +43,7 @@
 import static java.util.Collections.singletonList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
@@ -735,6 +737,26 @@ public void shouldNotBeFlushedAfterFlush() {
         assertTrue(producer.flushed());
     }
 
+    @Test
+    public void testMetadataOnException() throws InterruptedException {
+        buildMockProducer(false);
+        Future<RecordMetadata> metadata = producer.send(record2, (md, exception) -> {
+            assertNotNull(md);
+            assertEquals(md.offset(), -1L, "Invalid offset");
+            assertEquals(md.timestamp(), RecordBatch.NO_TIMESTAMP, "Invalid timestamp");
+            assertEquals(md.serializedKeySize(), -1L, "Invalid Serialized Key size");
+            assertEquals(md.serializedValueSize(), -1L, "Invalid Serialized value size");
+        });
+        IllegalArgumentException e = new IllegalArgumentException("dummy exception");
+        assertTrue(producer.errorNext(e), "Complete the second request with an error");
+        try {
+            metadata.get();
+            fail("Something went wrong, expected an error");
+        } catch (ExecutionException err) {
+            assertEquals(e, err.getCause());
+        }
+    }
+
     private boolean isError(Future<?> future) {
         try {
             future.get();