kafka/0011-ConfigEntry.patch

diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java b/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java
index 8f058a2d6c..8f3b336911 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/ConfigEntry.java
@@ -171,11 +171,13 @@ public class ConfigEntry {
         ConfigEntry that = (ConfigEntry) o;
 
         return this.name.equals(that.name) &&
-                this.value != null ? this.value.equals(that.value) : that.value == null &&
+                Objects.equals(this.value, that.value) &&
                 this.isSensitive == that.isSensitive &&
                 this.isReadOnly == that.isReadOnly &&
-                this.source == that.source &&
-                Objects.equals(this.synonyms, that.synonyms);
+                Objects.equals(this.source, that.source) &&
+                Objects.equals(this.synonyms, that.synonyms) &&
+                Objects.equals(this.type, that.type) &&
+                Objects.equals(this.documentation, that.documentation);
     }
 
     @Override
@@ -183,11 +185,13 @@ public class ConfigEntry {
         final int prime = 31;
         int result = 1;
         result = prime * result + name.hashCode();
-        result = prime * result + ((value == null) ? 0 : value.hashCode());
-        result = prime * result + (isSensitive ? 1 : 0);
+        result = prime * result + Objects.hashCode(value);
+	result = prime * result + (isSensitive ? 1 : 0);
         result = prime * result + (isReadOnly ? 1 : 0);
-        result = prime * result + source.hashCode();
-        result = prime * result + synonyms.hashCode();
+        result = prime * result + Objects.hashCode(source);
+        result = prime * result + Objects.hashCode(synonyms);
+        result = prime * result + Objects.hashCode(type);
+        result = prime * result + Objects.hashCode(documentation);
         return result;
     }
 
@@ -204,6 +208,8 @@ public class ConfigEntry {
                 ", isSensitive=" + isSensitive +
                 ", isReadOnly=" + isReadOnly +
                 ", synonyms=" + synonyms +
+                ", type=" + type +
+                ", documentation=" + documentation +
                 ")";
     }
 
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/ConfigTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/ConfigTest.java
index 4008a54382..59d1150ac3 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/ConfigTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/ConfigTest.java
@@ -82,4 +82,19 @@ public class ConfigTest {
                                              boolean isReadOnly, List<ConfigEntry.ConfigSynonym> synonyms) {
         return new ConfigEntry(name, value, source, isSensitive, isReadOnly, synonyms, ConfigType.UNKNOWN, null);
     }
+
+    @Test
+    public void testHashCodeAndEqualsWithNull() {
+        ConfigEntry ce0 = new ConfigEntry("abc", null, null, false, false, null, null, null);
+        ConfigEntry ce1 = new ConfigEntry("abc", null, null, false, false, null, null, null);
+        assertEquals(ce0, ce1);
+        assertEquals(ce0.hashCode(), ce1.hashCode());
+    }
+
+    @Test
+    public void testEquals() {
+        ConfigEntry ce0 = new ConfigEntry("abc", null, ConfigEntry.ConfigSource.DEFAULT_CONFIG, false, false, null, null, null);
+        ConfigEntry ce1 = new ConfigEntry("abc", null, ConfigEntry.ConfigSource.DYNAMIC_BROKER_CONFIG, false, false, null, null, null);
+        assertNotEquals(ce0, ce1);
+    }
 }
diff --git a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
index 8603fbec78..66b7fb1cc9 100644
--- a/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
+++ b/core/src/test/scala/unit/kafka/admin/ConfigCommandTest.scala
@@ -803,7 +803,14 @@ class ConfigCommandTest extends ZooKeeperTestHarness with Logging {
           new AlterConfigOp(newConfigEntry("min.insync.replicas", "2"), AlterConfigOp.OpType.SET),
           new AlterConfigOp(newConfigEntry("unclean.leader.election.enable", ""), AlterConfigOp.OpType.DELETE)
         )
-        assertEquals(expectedConfigOps, alterConfigOps.asScala.toSet)
+        assertEquals(expectedConfigOps.size, alterConfigOps.size)
+        expectedConfigOps.foreach { expectedOp =>
+          val actual = alterConfigOps.asScala.find(_.configEntry.name == expectedOp.configEntry.name)
+          assertNotEquals(actual, None)
+          assertEquals(expectedOp.opType, actual.get.opType)
+          assertEquals(expectedOp.configEntry.name, actual.get.configEntry.name)
+          assertEquals(expectedOp.configEntry.value, actual.get.configEntry.value)
+        }        
         alteredConfigs = true
         alterResult
       }