HOTFIX: Fix equality semantics of KeyValue

streams/src/main/java/org/apache/kafka/streams/KeyValue.java

@@ -63,22 +63,22 @@ public String toString() {
     }
 
     @Override
-    public boolean equals(Object other) {
-        if (this == other)
+    public boolean equals(Object obj) {
+        if (this == obj)
             return true;
 
-        if (other instanceof KeyValue) {
-            KeyValue otherKV = (KeyValue) other;
-
-            return key == null ? otherKV.key == null : key.equals(otherKV.key)
-                    && value == null ? otherKV.value == null : value.equals(otherKV.value);
-        } else {
+        if (!(obj instanceof KeyValue)) {
             return false;
         }
+
+        KeyValue other = (KeyValue) obj;
+        return (this.key == null ? other.key == null : this.key.equals(other.key))
+                && (this.value == null ? other.value == null : this.value.equals(other.value));
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(key, value);
     }
+
 }

streams/src/test/java/org/apache/kafka/streams/KeyValueTest.java

@@ -24,27 +24,50 @@
 
 public class KeyValueTest {
 
-    private KeyValue<String, Long> kv1a = new KeyValue<>("key1", 1L);
-    private KeyValue<String, Long> kv1b = new KeyValue<>("key1", 1L);
-    private KeyValue<String, Long> kv2 = new KeyValue<>("key2", 2L);
-    private KeyValue<String, Long> kv3 = new KeyValue<>("key3", 3L);
-
     @Test
-    public void testEquals() {
-        assertTrue(kv1a.equals(kv1a));
-        assertTrue(kv1a.equals(kv1b));
-        assertTrue(kv1b.equals(kv1a));
-        assertFalse(kv1a.equals(kv2));
-        assertFalse(kv1a.equals(kv3));
-        assertFalse(kv2.equals(kv3));
-        assertFalse(kv1a.equals(null));
-    }
+    public void shouldHaveSaneEqualsAndHashCode() {
+        KeyValue<String, Long> kv = KeyValue.pair("key1", 1L);
+        KeyValue<String, Long> copyOfKV = KeyValue.pair(kv.key, kv.value);
 
-    @Test
-    public void testHashcode() {
-        assertTrue(kv1a.hashCode() == kv1b.hashCode());
-        assertFalse(kv1a.hashCode() == kv2.hashCode());
-        assertFalse(kv1a.hashCode() == kv3.hashCode());
-        assertFalse(kv2.hashCode() == kv3.hashCode());
+        // Reflexive
+        assertTrue(kv.equals(kv));
+        assertTrue(kv.hashCode() == kv.hashCode());
+
+        // Symmetric
+        assertTrue(kv.equals(copyOfKV));
+        assertTrue(kv.hashCode() == copyOfKV.hashCode());
+        assertTrue(copyOfKV.hashCode() == kv.hashCode());
+
+        // Transitive
+        KeyValue<String, Long> copyOfCopyOfKV = KeyValue.pair(copyOfKV.key, copyOfKV.value);
+        assertTrue(copyOfKV.equals(copyOfCopyOfKV));
+        assertTrue(copyOfKV.hashCode() == copyOfCopyOfKV.hashCode());
+        assertTrue(kv.equals(copyOfCopyOfKV));
+        assertTrue(kv.hashCode() == copyOfCopyOfKV.hashCode());
+
+        // Inequality scenarios
+        assertFalse("must be false for null", kv.equals(null));
+        assertFalse("must be false if key is non-null and other key is null", kv.equals(KeyValue.pair(null, kv.value)));
+        assertFalse("must be false if value is non-null and other value is null", kv.equals(KeyValue.pair(kv.key, null)));
+        KeyValue<Long, Long> differentKeyType = KeyValue.pair(1L, kv.value);
+        assertFalse("must be false for different key types", kv.equals(differentKeyType));
+        KeyValue<String, String> differentValueType = KeyValue.pair(kv.key, "anyString");
+        assertFalse("must be false for different value types", kv.equals(differentValueType));
+        KeyValue<Long, String> differentKeyValueTypes = KeyValue.pair(1L, "anyString");
+        assertFalse("must be false for different key and value types", kv.equals(differentKeyValueTypes));
+        assertFalse("must be false for different types of objects", kv.equals(new Object()));
+
+        KeyValue<String, Long> differentKey = KeyValue.pair(kv.key + "suffix", kv.value);
+        assertFalse("must be false if key is different", kv.equals(differentKey));
+        assertFalse("must be false if key is different", differentKey.equals(kv));
+
+        KeyValue<String, Long> differentValue = KeyValue.pair(kv.key, kv.value + 1L);
+        assertFalse("must be false if value is different", kv.equals(differentValue));
+        assertFalse("must be false if value is different", differentValue.equals(kv));
+
+        KeyValue<String, Long> differentKeyAndValue = KeyValue.pair(kv.key + "suffix", kv.value + 1L);
+        assertFalse("must be false if key and value are different", kv.equals(differentKeyAndValue));
+        assertFalse("must be false if key and value are different", differentKeyAndValue.equals(kv));
     }
-}
+
+}