tmp

Signed-off-by: PoAn Yang <[email protected]>

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

@@ -543,7 +543,7 @@ public void transitionToJoining() {
      * to leave the group has been sent out.
      */
     public CompletableFuture<Void> leaveGroup() {
-        if (isNotInGroup()) {
+        if (isNotInGroup() || state == MemberState.JOINING) {
             if (state == MemberState.FENCED) {
                 clearAssignment();
                 transitionTo(MemberState.UNSUBSCRIBED);

clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java

@@ -252,7 +252,9 @@ private void process(final SubscriptionChangeEvent ignored) {
      */
     private void process(final UnsubscribeEvent event) {
         if (requestManagers.consumerHeartbeatRequestManager.isPresent()) {
+            System.out.println("UnsubscribeEvent: " + event);
             CompletableFuture<Void> future = requestManagers.consumerHeartbeatRequestManager.get().membershipManager().leaveGroup();
+            System.out.println("UnsubscribeEvent: " + future.isCompletedExceptionally());
             future.whenComplete(complete(event.future()));
         } else {
             // If the consumer is not using the group management capabilities, we still need to clear all assignments it may have.

core/src/test/scala/integration/kafka/api/PlaintextConsumerSubscriptionTest.scala

@@ -244,6 +244,7 @@ class PlaintextConsumerSubscriptionTest extends AbstractConsumerTest {
     }, waitTimeMs = 5000, msg = "An InvalidTopicException should be thrown.")
 
     assertEquals(s"Invalid topics: [${invalidTopicName}]", exception.getMessage)
+    Thread.sleep(1000)
   }
 }