MINOR: fix a bug in removing elements from an ImplicitLinkedHashColle…

clients/src/main/java/org/apache/kafka/common/utils/ImplicitLinkedHashCollection.java

@@ -208,8 +208,7 @@ public void remove() {
                 throw new IllegalStateException();
             }
             Element nextElement = indexToElement(head, elements, lastReturned.next());
-            removeFromList(head, elements, nextElement.prev());
-            size--;
+            ImplicitLinkedHashCollection.this.removeElementAtSlot(nextElement.prev());
             if (lastReturned == cur) {
                 // If the element we are removing was cur, set cur to cur->next.
                 cur = nextElement;

clients/src/test/java/org/apache/kafka/common/utils/ImplicitLinkedHashCollectionTest.java

@@ -103,7 +103,8 @@ public String toString() {
 
         @Override
         public int hashCode() {
-            return key;
+            long hashCode = 2654435761L * key;
+            return (int) (hashCode >> 32);
         }
     }
 
@@ -596,4 +597,25 @@ public void testMoveToEnd() {
         Assert.assertThrows(RuntimeException.class, () ->
             coll.moveToEnd(new TestElement(4, 4)));
     }
+
+    @Test
+    public void testRemovals() {
+        ImplicitLinkedHashCollection<TestElement> coll = new ImplicitLinkedHashCollection<>();
+        List<TestElement> elements  = new ArrayList<>();
+        for (int i = 0; i < 100; i++) {
+            TestElement element  = new TestElement(i, i);
+            elements.add(element);
+            coll.add(element);
+        }
+        assertEquals(100, coll.size());
+        Iterator<TestElement> iter = coll.iterator();
+        for (int i = 0; i < 50; i++) {
+            iter.next();
+            iter.remove();
+        }
+        assertEquals(50, coll.size());
+        for (int i = 50; i < 100; i++) {
+            assertEquals(new TestElement(i, i), coll.find(elements.get(i)));
+        }
+    }
 }