KAFKA-9288: Do not allow the same object to be inserted multiple time…

…s into ImplicitLinkedHashCollection (apache#7809) Reviewers: Jason Gustafson <[email protected]>
agam · Dec 11, 2019 · 4fea3e4 · 4fea3e4
1 parent f770b49
commit 4fea3e4
Show file tree

Hide file tree

Showing 4 changed files with 36 additions and 2 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/ImplicitLinkedHashCollection.java b/clients/src/main/java/org/apache/kafka/common/utils/ImplicitLinkedHashCollection.java
@@ -380,6 +380,9 @@ final public boolean add(E newElement) {
         if (newElement == null) {
             return false;
         }
+        if (newElement.prev() != INVALID_INDEX || newElement.next() != INVALID_INDEX) {
+            return false;
+        }
         if ((size + 1) >= elements.length / 2) {
             changeCapacity(calculateCapacity(elements.length));
         }

diff --git a/clients/src/test/java/org/apache/kafka/common/utils/ImplicitLinkedHashCollectionTest.java b/clients/src/test/java/org/apache/kafka/common/utils/ImplicitLinkedHashCollectionTest.java
@@ -492,6 +492,19 @@ public void testManyInsertsAndDeletes() {
         }
     }
 
+    @Test
+    public void testInsertingTheSameObjectMultipleTimes() {
+        ImplicitLinkedHashCollection<TestElement> coll = new ImplicitLinkedHashCollection<>();
+        TestElement element = new TestElement(123);
+        assertTrue(coll.add(element));
+        assertFalse(coll.add(element));
+        assertFalse(coll.add(element));
+        assertTrue(coll.remove(element));
+        assertFalse(coll.remove(element));
+        assertTrue(coll.add(element));
+        assertFalse(coll.add(element));
+    }
+
     @Test
     public void testEquals() {
         ImplicitLinkedHashCollection<TestElement> coll1 = new ImplicitLinkedHashCollection<>();

diff --git a/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java
@@ -80,7 +80,7 @@ private void generateClass(Optional<MessageSpec> topLevelMessageSpec,
         generateFieldDeclarations(struct, isSetElement);
         buffer.printf("%n");
         schemaGenerator.writeSchema(className, buffer);
-        generateClassConstructors(className, struct);
+        generateClassConstructors(className, struct, isSetElement);
         buffer.printf("%n");
         if (isTopLevel) {
             generateShortAccessor("apiKey", topLevelMessageSpec.get().apiKey().orElse((short) -1));
@@ -385,18 +385,28 @@ private String fieldConcreteJavaType(FieldSpec field) {
         }
     }
 
-    private void generateClassConstructors(String className, StructSpec struct) {
+    private void generateClassConstructors(String className, StructSpec struct, boolean isSetElement) {
         headerGenerator.addImport(MessageGenerator.READABLE_CLASS);
         buffer.printf("public %s(Readable _readable, short _version) {%n", className);
         buffer.incrementIndent();
         buffer.printf("read(_readable, _version);%n");
+        if (isSetElement) {
+            headerGenerator.addImport(MessageGenerator.IMPLICIT_LINKED_HASH_COLLECTION_CLASS);
+            buffer.printf("this.prev = ImplicitLinkedHashCollection.INVALID_INDEX;%n");
+            buffer.printf("this.next = ImplicitLinkedHashCollection.INVALID_INDEX;%n");
+        }
         buffer.decrementIndent();
         buffer.printf("}%n");
         buffer.printf("%n");
         headerGenerator.addImport(MessageGenerator.STRUCT_CLASS);
         buffer.printf("public %s(Struct struct, short _version) {%n", className);
         buffer.incrementIndent();
         buffer.printf("fromStruct(struct, _version);%n");
+        if (isSetElement) {
+            headerGenerator.addImport(MessageGenerator.IMPLICIT_LINKED_HASH_COLLECTION_CLASS);
+            buffer.printf("this.prev = ImplicitLinkedHashCollection.INVALID_INDEX;%n");
+            buffer.printf("this.next = ImplicitLinkedHashCollection.INVALID_INDEX;%n");
+        }
         buffer.decrementIndent();
         buffer.printf("}%n");
         buffer.printf("%n");
@@ -406,6 +416,11 @@ private void generateClassConstructors(String className, StructSpec struct) {
             buffer.printf("this.%s = %s;%n",
                 field.camelCaseName(), fieldDefault(field));
         }
+        if (isSetElement) {
+            headerGenerator.addImport(MessageGenerator.IMPLICIT_LINKED_HASH_COLLECTION_CLASS);
+            buffer.printf("this.prev = ImplicitLinkedHashCollection.INVALID_INDEX;%n");
+            buffer.printf("this.next = ImplicitLinkedHashCollection.INVALID_INDEX;%n");
+        }
         buffer.decrementIndent();
         buffer.printf("}%n");
     }

diff --git a/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java b/generator/src/main/java/org/apache/kafka/message/MessageGenerator.java
@@ -61,6 +61,9 @@ public final class MessageGenerator {
 
     static final String ARRAYLIST_CLASS = "java.util.ArrayList";
 
+    static final String IMPLICIT_LINKED_HASH_COLLECTION_CLASS =
+        "org.apache.kafka.common.utils.ImplicitLinkedHashCollection";
+
     static final String IMPLICIT_LINKED_HASH_MULTI_COLLECTION_CLASS =
         "org.apache.kafka.common.utils.ImplicitLinkedHashMultiCollection";