Skip to content

Commit 1c5228c

Browse files
pranavrthemasab
pranavrth
authored and
emasab
committed
[KIP-848] HB Error Code, Partial ack flow, OffsetCommit Request, Response and various fixes (#4634)
- Added error handling to ConsumerGroupHeartbeat API - Added type new errors - UNRELEASED_INSTANCE_ID and UNSUPPORTED_ASSIGNOR - Added partial acknowledgement flow - Upgraded OffsetCommit Request and response to v9 - Fixed metadata being called with duplicate topic id - Fixed next_target_assignment not getting reset to NULL - Fixed member stuck if fenced during rebalancing - Fixed segfault with current and target assignment while resetting consumer group - Fixed segfault due to deleted topic in metadata - Fixed leave not being called if the consumer without any assignment leaves
1 parent 0c545a5 commit 1c5228c

16 files changed

+491
-223
lines changed

‎INTRODUCTION.md

Lines changed: 40 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -1967,50 +1967,50 @@ The [Apache Kafka Implementation Proposals (KIPs)](https://cwiki.apache.org/conf
19671967

19681968
### Supported protocol versions
19691969

1970-
"Kafka max" is the maximum ApiVersion supported in Apache Kafka 3.5.0, while
1970+
"Kafka max" is the maximum ApiVersion supported in Apache Kafka 3.7.0, while
19711971
"librdkafka max" is the maximum ApiVersion supported in the latest
19721972
release of librdkafka.
19731973

19741974

1975-
| ApiKey | Request name | Kafka max | librdkafka max |
1976-
| ------- | ------------------------------| ----------- | ----------------------- |
1977-
| 0 | Produce | 10 | 8 |
1978-
| 1 | Fetch | 15 | 11 |
1979-
| 2 | ListOffsets | 8 | 7 |
1980-
| 3 | Metadata | 12 | 12 |
1981-
| 8 | OffsetCommit | 8 | 7 |
1982-
| 9 | OffsetFetch | 8 | 7 |
1983-
| 10 | FindCoordinator | 4 | 2 |
1984-
| 11 | JoinGroup | 9 | 5 |
1985-
| 12 | Heartbeat | 4 | 3 |
1986-
| 13 | LeaveGroup | 5 | 1 |
1987-
| 14 | SyncGroup | 5 | 3 |
1988-
| 15 | DescribeGroups | 5 | 4 |
1989-
| 16 | ListGroups | 4 | 4 |
1990-
| 17 | SaslHandshake | 1 | 1 |
1991-
| 18 | ApiVersions | 3 | 3 |
1992-
| 19 | CreateTopics | 7 | 4 |
1993-
| 20 | DeleteTopics | 6 | 1 |
1994-
| 21 | DeleteRecords | 2 | 1 |
1995-
| 22 | InitProducerId | 4 | 4 |
1996-
| 23 | OffsetForLeaderEpoch | 4 | 2 |
1997-
| 24 | AddPartitionsToTxn | 4 | 0 |
1998-
| 25 | AddOffsetsToTxn | 3 | 0 |
1999-
| 26 | EndTxn | 3 | 1 |
2000-
| 28 | TxnOffsetCommit | 3 | 3 |
2001-
| 29 | DescribeAcls | 3 | 1 |
2002-
| 30 | CreateAcls | 3 | 1 |
2003-
| 31 | DeleteAcls | 3 | 1 |
2004-
| 32 | DescribeConfigs | 4 | 1 |
2005-
| 33 | AlterConfigs | 2 | 2 |
2006-
| 36 | SaslAuthenticate | 2 | 1 |
2007-
| 37 | CreatePartitions | 3 | 0 |
2008-
| 42 | DeleteGroups | 2 | 1 |
2009-
| 44 | IncrementalAlterConfigs | 1 | 1 |
2010-
| 47 | OffsetDelete | 0 | 0 |
2011-
| 50 | DescribeUserScramCredentials | 0 | 0 |
2012-
| 51 | AlterUserScramCredentials | 0 | 0 |
2013-
1975+
| ApiKey | Request name | Kafka max | librdkafka max |
1976+
| ------- | ----------------------------- | ---------- | -------------- |
1977+
| 0 | Produce | 10 | 8 |
1978+
| 1 | Fetch | 16 | 11 |
1979+
| 2 | ListOffsets | 8 | 7 |
1980+
| 3 | Metadata | 12 | 12 |
1981+
| 8 | OffsetCommit | 9 | 9 |
1982+
| 9 | OffsetFetch | 9 | 9 |
1983+
| 10 | FindCoordinator | 4 | 2 |
1984+
| 11 | JoinGroup | 9 | 5 |
1985+
| 12 | Heartbeat | 4 | 3 |
1986+
| 13 | LeaveGroup | 5 | 1 |
1987+
| 14 | SyncGroup | 5 | 3 |
1988+
| 15 | DescribeGroups | 5 | 4 |
1989+
| 16 | ListGroups | 4 | 4 |
1990+
| 17 | SaslHandshake | 1 | 1 |
1991+
| 18 | ApiVersions | 3 | 3 |
1992+
| 19 | CreateTopics | 7 | 4 |
1993+
| 20 | DeleteTopics | 6 | 1 |
1994+
| 21 | DeleteRecords | 2 | 1 |
1995+
| 22 | InitProducerId | 4 | 4 |
1996+
| 23 | OffsetForLeaderEpoch | 4 | 2 |
1997+
| 24 | AddPartitionsToTxn | 4 | 0 |
1998+
| 25 | AddOffsetsToTxn | 3 | 0 |
1999+
| 26 | EndTxn | 3 | 1 |
2000+
| 28 | TxnOffsetCommit | 3 | 3 |
2001+
| 29 | DescribeAcls | 3 | 1 |
2002+
| 30 | CreateAcls | 3 | 1 |
2003+
| 31 | DeleteAcls | 3 | 1 |
2004+
| 32 | DescribeConfigs | 4 | 1 |
2005+
| 33 | AlterConfigs | 2 | 2 |
2006+
| 36 | SaslAuthenticate | 2 | 1 |
2007+
| 37 | CreatePartitions | 3 | 0 |
2008+
| 42 | DeleteGroups | 2 | 1 |
2009+
| 44 | IncrementalAlterConfigs | 1 | 1 |
2010+
| 47 | OffsetDelete | 0 | 0 |
2011+
| 50 | DescribeUserScramCredentials | 0 | 0 |
2012+
| 51 | AlterUserScramCredentials | 0 | 0 |
2013+
| 68 | ConsumerGroupHeartbeat | 0 | 0 |
20142014

20152015
# Recommendations for language binding developers
20162016

‎examples/consumer.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -258,4 +258,4 @@ int main(int argc, char **argv) {
258258
rd_kafka_destroy(rk);
259259

260260
return 0;
261-
}
261+
}

‎src/rdkafka.c

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -707,6 +707,12 @@ static const struct rd_kafka_err_desc rd_kafka_err_descs[] = {
707707
_ERR_DESC(RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_ID, "Broker: Unknown topic id"),
708708
_ERR_DESC(RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH,
709709
"Broker: The member epoch is fenced by the group coordinator"),
710+
_ERR_DESC(RD_KAFKA_RESP_ERR_UNRELEASED_INSTANCE_ID,
711+
"Broker: The instance ID is still used by another member in the "
712+
"consumer group"),
713+
_ERR_DESC(RD_KAFKA_RESP_ERR_UNSUPPORTED_ASSIGNOR,
714+
"Broker: The assignor or its version range is not supported by "
715+
"the consumer group"),
710716
_ERR_DESC(RD_KAFKA_RESP_ERR_STALE_MEMBER_EPOCH,
711717
"Broker: The member epoch is stale"),
712718
_ERR_DESC(RD_KAFKA_RESP_ERR__END, NULL)};

‎src/rdkafka.h

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -638,6 +638,12 @@ typedef enum {
638638
RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_ID = 100,
639639
/** The member epoch is fenced by the group coordinator */
640640
RD_KAFKA_RESP_ERR_FENCED_MEMBER_EPOCH = 110,
641+
/** The instance ID is still used by another member in the
642+
* consumer group */
643+
RD_KAFKA_RESP_ERR_UNRELEASED_INSTANCE_ID = 111,
644+
/** The assignor or its version range is not supported by the consumer
645+
* group */
646+
RD_KAFKA_RESP_ERR_UNSUPPORTED_ASSIGNOR = 112,
641647
/** The member epoch is stale */
642648
RD_KAFKA_RESP_ERR_STALE_MEMBER_EPOCH = 113,
643649
RD_KAFKA_RESP_ERR_END_ALL,

0 commit comments

Comments
 (0)