Fix hang in incremental assign during close (#4528)

While using the cooperative assignor, given an assignment is received while closing the consumer it's possible that it gets stuck in state WAIT_ASSIGN_CALL, while the method is converted to a full unassign. Solved by changing state from WAIT_ASSIGN_CALL to WAIT_UNASSIGN_CALL while doing this conversion. Closes #4527
confluentinc · Jun 10, 2024 · 8c64f5d · 8c64f5d
1 parent 3536f56
commit 8c64f5d
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 2 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -20,6 +20,8 @@ librdkafka v2.4.0 is a feature release:
    a consumer group (#4678).
  * Fix to metadata refresh interruption (#4679).
  * Fix for an undesired partition migration with stale leader epoch (#4680).
+ * Fix hang in cooperative consumer mode if an assignment is processed
+   while closing the consumer (#4528).
 
 
 ## Upgrade considerations
@@ -79,6 +81,12 @@ librdkafka v2.4.0 is a feature release:
    Fixed by ensuring that a resume operation is completely a no-op when
    the partition isn't paused.
    Happening since 1.x (#4636).
+ * Issues: #4527.
+   While using the cooperative assignor, given an assignment is received while closing the consumer
+   it's possible that it gets stuck in state WAIT_ASSIGN_CALL, while the method is converted to
+   a full unassign. Solved by changing state from WAIT_ASSIGN_CALL to WAIT_UNASSIGN_CALL
+   while doing this conversion.
+   Happening since 1.x (#4528).
 
 
 

diff --git a/tests/test.c b/tests/test.c
@@ -7307,7 +7307,7 @@ const char *test_consumer_group_protocol() {
 
 int test_consumer_group_protocol_generic() {
         return !test_consumer_group_protocol_str ||
-               !strcmp(test_consumer_group_protocol_str, "generic");
+               !strcmp(test_consumer_group_protocol_str, "classic");
 }
 
 int test_consumer_group_protocol_consumer() {

diff --git a/tests/test.h b/tests/test.h
@@ -246,7 +246,7 @@ static RD_INLINE RD_UNUSED void rtrim(char *str) {
                 return RET;                                                    \
         }                                                                      \
         if (test_consumer_group_protocol() &&                                  \
-            strcmp(test_consumer_group_protocol(), "generic")) {               \
+            strcmp(test_consumer_group_protocol(), "classic")) {               \
                 TEST_SKIP(                                                     \
                     "Mock cluster cannot be used "                             \
                     "with group.protocol=%s\n",                                \