Defer commit on transport error to avoid consumer_close hang

confluentinc · Aug 1, 2019 · 7af5c3d · 7af5c3d
1 parent eb11929
commit 7af5c3d
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/src/rdkafka_cgrp.c b/src/rdkafka_cgrp.c
@@ -1902,7 +1902,10 @@ static void rd_kafka_cgrp_op_handle_OffsetCommit (rd_kafka_t *rk,
         if (err == RD_KAFKA_RESP_ERR__IN_PROGRESS)
                 return; /* Retrying */
         else if (err == RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP ||
-                 err == RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE) {
+                 err == RD_KAFKA_RESP_ERR_GROUP_COORDINATOR_NOT_AVAILABLE ||
+                 err == RD_KAFKA_RESP_ERR__TRANSPORT) {
+                /* The coordinator is not available, defer the offset commit
+                 * to when the coordinator is back up again. */
 
                 /* future-proofing, see timeout_scan(). */
                 rd_kafka_assert(NULL, err != RD_KAFKA_RESP_ERR__WAIT_COORD);

diff --git a/src/rdkafka_request.c b/src/rdkafka_request.c
@@ -798,6 +798,9 @@ rd_kafka_handle_OffsetCommit (rd_kafka_t *rk,
 		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_SPECIAL,
 		RD_KAFKA_RESP_ERR_NOT_COORDINATOR_FOR_GROUP,
 
+                RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_SPECIAL,
+                RD_KAFKA_RESP_ERR__TRANSPORT,
+
 		RD_KAFKA_ERR_ACTION_REFRESH|RD_KAFKA_ERR_ACTION_RETRY,
 		RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION,