KAFKA-1805; ProducerRecord should implement equals and hashCode; reviewed by Guozhang Wang

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

@@ -99,4 +99,28 @@ public final class ProducerRecord<K, V> {
         String value = this.value == null ? "null" : this.value.toString();
         return "ProducerRecord(topic=" + topic + ", partition=" + partition + ", key=" + key + ", value=" + value;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof ProducerRecord)) return false;
+
+        ProducerRecord that = (ProducerRecord) o;
+
+        if (key != null ? !key.equals(that.key) : that.key != null) return false;
+        if (partition != null ? !partition.equals(that.partition) : that.partition != null) return false;
+        if (topic != null ? !topic.equals(that.topic) : that.topic != null) return false;
+        if (value != null ? !value.equals(that.value) : that.value != null) return false;
+
+        return true;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = topic != null ? topic.hashCode() : 0;
+        result = 31 * result + (partition != null ? partition.hashCode() : 0);
+        result = 31 * result + (key != null ? key.hashCode() : 0);
+        result = 31 * result + (value != null ? value.hashCode() : 0);
+        return result;
+    }
 }

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

@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.producer;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class ProducerRecordTest {
+
+    @Test
+    public void testEqualsAndHashCode() {
+        ProducerRecord<String, Integer> producerRecord = new ProducerRecord<String, Integer>("test", 1 , "key", 1);
+        assertEquals(producerRecord, producerRecord);
+        assertEquals(producerRecord.hashCode(), producerRecord.hashCode());
+
+        ProducerRecord<String, Integer> equalRecord = new ProducerRecord<String, Integer>("test", 1 , "key", 1);
+        assertEquals(producerRecord, equalRecord);
+        assertEquals(producerRecord.hashCode(), equalRecord.hashCode());
+
+        ProducerRecord<String, Integer> topicMisMatch = new ProducerRecord<String, Integer>("test-1", 1 , "key", 1);
+        assertFalse(producerRecord.equals(topicMisMatch));
+
+        ProducerRecord<String, Integer> partitionMismatch = new ProducerRecord<String, Integer>("test", 2 , "key", 1);
+        assertFalse(producerRecord.equals(partitionMismatch));
+
+        ProducerRecord<String, Integer> keyMisMatch = new ProducerRecord<String, Integer>("test", 1 , "key-1", 1);
+        assertFalse(producerRecord.equals(keyMisMatch));
+
+        ProducerRecord<String, Integer> valueMisMatch = new ProducerRecord<String, Integer>("test", 1 , "key", 2);
+        assertFalse(producerRecord.equals(valueMisMatch));
+
+        ProducerRecord<String, Integer> nullFieldsRecord = new ProducerRecord<String, Integer>("topic", null, null, null);
+        assertEquals(nullFieldsRecord, nullFieldsRecord);
+        assertEquals(nullFieldsRecord.hashCode(), nullFieldsRecord.hashCode());
+    }
+}