Check for duplicate state listener

Author: cadonna

clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManager.java

@@ -354,7 +354,13 @@ private boolean isNotInGroup() {
      * @param listener Listener to invoke.
      */
     public void registerStateListener(MemberStateListener listener) {
-        stateUpdatesListeners.add(Objects.requireNonNull(listener, "State updates listener cannot be null"));
+        Objects.requireNonNull(listener, "State updates listener cannot be null");
+        for (MemberStateListener registeredListener : stateUpdatesListeners) {
+            if (registeredListener == listener) {
+                throw new IllegalArgumentException("Listener is already registered.");
+            }
+        }
+        stateUpdatesListeners.add(listener);
     }
 
     /**

clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsMembershipManagerTest.java

@@ -1458,6 +1458,28 @@ public void testMaybeRejoinStaleMember() {
         assertEquals(StreamsGroupHeartbeatRequest.JOIN_GROUP_MEMBER_EPOCH, membershipManager.memberEpoch());
     }
 
+    @Test
+    public void testForDuplicateRegistrationOfSameStateListener() {
+        final MemberStateListener listener1 = new MemberStateListener() {
+
+            @Override
+            public void onMemberEpochUpdated(Optional<Integer> memberEpoch, String memberId) {
+            }
+        };
+        final MemberStateListener listener2 = new MemberStateListener() {
+
+            @Override
+            public void onMemberEpochUpdated(Optional<Integer> memberEpoch, String memberId) {
+            }
+        };
+
+        membershipManager.registerStateListener(listener1);
+        membershipManager.registerStateListener(listener2);
+        final Exception exception =
+            assertThrows(IllegalArgumentException.class, () -> membershipManager.registerStateListener(listener1));
+        assertEquals("Listener is already registered.", exception.getMessage());
+    }
+
     private void verifyThatNoTasksHaveBeenRevoked() {
         verify(streamsRebalanceEventsProcessor, never()).requestOnTasksRevokedCallbackInvocation(any());
         verify(subscriptionState, never()).markPendingRevocation(any());