Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Produce upgrade to flexver plus tests #4595

Merged
merged 6 commits into from
Jun 13, 2024
Merged

Produce upgrade to flexver plus tests #4595

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
55799e4
Produce flexver upgrade
emasab Dec 15, 2023
2c9fd19
Fix response parsing
anchitj Jan 16, 2024
36cb617
Make rd_kafka_buf_skip_str generic
anchitj May 17, 2024
d99a67a
PR feedback
anchitj Jun 10, 2024
9c73607
PR feedback
anchitj Jun 11, 2024
1224b57
PR feedback
anchitj Jun 13, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion INTRODUCTION.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2068,7 +2068,7 @@ release of librdkafka.

| ApiKey | Request name | Kafka max | librdkafka max |
| ------- | ----------------------------- | ---------- | -------------- |
| 0 | Produce | 10 | 8 |
| 0 | Produce | 10 | 9 |
| 1 | Fetch | 16 | 15 |
| 2 | ListOffsets | 8 | 7 |
| 3 | Metadata | 12 | 12 |
Expand Down
18 changes: 16 additions & 2 deletions src/rdkafka_buf.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -719,15 +719,29 @@ struct rd_kafka_buf_s { /* rd_kafka_buf_t */
} while (0)

/**
* Skip a string.
* Skip a string without flexver.
*/
#define rd_kafka_buf_skip_str(rkbuf) \
#define rd_kafka_buf_skip_str_no_flexver(rkbuf) \
do { \
int16_t _slen; \
rd_kafka_buf_read_i16(rkbuf, &_slen); \
rd_kafka_buf_skip(rkbuf, RD_KAFKAP_STR_LEN0(_slen)); \
} while (0)

/**
* Skip a string (generic).
*/
#define rd_kafka_buf_skip_str(rkbuf) \
do { \
if ((rkbuf)->rkbuf_flags & RD_KAFKA_OP_F_FLEXVER) { \
uint64_t _uva; \
rd_kafka_buf_read_uvarint(rkbuf, &_uva); \
rd_kafka_buf_skip( \
rkbuf, RD_KAFKAP_STR_LEN0(((int64_t)_uva) - 1)); \
} else { \
rd_kafka_buf_skip_str_no_flexver(rkbuf); \
} \
} while (0)
/**
* Read Kafka COMPACT_BYTES representation (VARINT+N) or
* standard BYTES representation(4+N).
Expand Down
2 changes: 1 addition & 1 deletion src/rdkafka_mock.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1106,7 +1106,7 @@ rd_kafka_mock_connection_read_request(rd_kafka_mock_connection_t *mconn,
RD_KAFKAP_REQHDR_SIZE);

/* For convenience, shave off the ClientId */
rd_kafka_buf_skip_str(rkbuf);
rd_kafka_buf_skip_str_no_flexver(rkbuf);

/* And the flexible versions header tags, if any */
rd_kafka_buf_skip_tags(rkbuf);
Expand Down
37 changes: 31 additions & 6 deletions src/rdkafka_mock_handlers.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,10 +61,11 @@ static int rd_kafka_mock_handle_Produce(rd_kafka_mock_connection_t *mconn,

rd_kafka_buf_read_i16(rkbuf, &Acks);
rd_kafka_buf_read_i32(rkbuf, &TimeoutMs);
rd_kafka_buf_read_i32(rkbuf, &TopicsCnt);
/* #Topics */
rd_kafka_buf_read_arraycnt(rkbuf, &TopicsCnt, RD_KAFKAP_TOPICS_MAX);
anchitj marked this conversation as resolved.
Show resolved Hide resolved

/* Response: #Topics */
rd_kafka_buf_write_i32(resp, TopicsCnt);
rd_kafka_buf_write_arraycnt(resp, TopicsCnt);

/* Inject error, if any */
all_err = rd_kafka_mock_next_request_error(mconn, resp);
Expand All @@ -75,14 +76,15 @@ static int rd_kafka_mock_handle_Produce(rd_kafka_mock_connection_t *mconn,
rd_kafka_mock_topic_t *mtopic;

rd_kafka_buf_read_str(rkbuf, &Topic);
rd_kafka_buf_read_i32(rkbuf, &PartitionCnt);
rd_kafka_buf_read_arraycnt(rkbuf, &PartitionCnt,
RD_KAFKAP_PARTITIONS_MAX);

mtopic = rd_kafka_mock_topic_find_by_kstr(mcluster, &Topic);

/* Response: Topic */
rd_kafka_buf_write_kstr(resp, &Topic);
/* Response: #Partitions */
rd_kafka_buf_write_i32(resp, PartitionCnt);
rd_kafka_buf_write_arraycnt(resp, PartitionCnt);

while (PartitionCnt-- > 0) {
int32_t Partition;
Expand All @@ -99,6 +101,9 @@ static int rd_kafka_mock_handle_Produce(rd_kafka_mock_connection_t *mconn,

rd_kafka_buf_read_kbytes(rkbuf, &records);

/* Partition Tags */
rd_kafka_buf_skip_tags(rkbuf);

/* Response: Partition */
rd_kafka_buf_write_i32(resp, Partition);

Expand Down Expand Up @@ -146,15 +151,35 @@ static int rd_kafka_mock_handle_Produce(rd_kafka_mock_connection_t *mconn,
resp, mpart->start_offset);
}
}

if (rkbuf->rkbuf_reqhdr.ApiVersion >= 8) {
/* Response: #RecordErrors
* TODO: Add support for injecting RecordErrors
* 0 record errors for now */
rd_kafka_buf_write_arraycnt(resp, 0);

/* Response: ErrorMessage */
rd_kafka_buf_write_str(resp, NULL, 0);
}
/* Response: Partition tags */
rd_kafka_buf_write_tags_empty(resp);
}

/* Topic tags */
rd_kafka_buf_skip_tags(rkbuf);
/* Response: Topic tags */
rd_kafka_buf_write_tags_empty(resp);
}

if (rkbuf->rkbuf_reqhdr.ApiVersion >= 1) {
/* Response: ThrottleTime */
rd_kafka_buf_write_i32(resp, 0);
}

rd_kafka_mock_connection_send_response(mconn, resp);
/* Response: Top level tags */
rd_kafka_buf_write_tags_empty(resp);

rd_kafka_mock_connection_send_response0(mconn, resp, rd_true);
anchitj marked this conversation as resolved.
Show resolved Hide resolved

return 0;

Expand Down Expand Up @@ -2281,7 +2306,7 @@ rd_kafka_mock_handle_OffsetForLeaderEpoch(rd_kafka_mock_connection_t *mconn,
const struct rd_kafka_mock_api_handler
rd_kafka_mock_api_handlers[RD_KAFKAP__NUM] = {
/* [request-type] = { MinVersion, MaxVersion, FlexVersion, callback } */
[RD_KAFKAP_Produce] = {0, 7, -1, rd_kafka_mock_handle_Produce},
[RD_KAFKAP_Produce] = {0, 9, 9, rd_kafka_mock_handle_Produce},
[RD_KAFKAP_Fetch] = {0, 15, 12, rd_kafka_mock_handle_Fetch},
[RD_KAFKAP_ListOffsets] = {0, 7, 6, rd_kafka_mock_handle_ListOffsets},
[RD_KAFKAP_OffsetFetch] = {0, 6, 6, rd_kafka_mock_handle_OffsetFetch},
Expand Down
27 changes: 17 additions & 10 deletions src/rdkafka_msgset_writer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@


/** @brief The maxium ProduceRequestion ApiVersion supported by librdkafka */
static const int16_t rd_kafka_ProduceRequest_max_version = 8;
static const int16_t rd_kafka_ProduceRequest_max_version = 9;


typedef struct rd_kafka_msgset_writer_s {
Expand Down Expand Up @@ -267,6 +267,7 @@ static void rd_kafka_msgset_writer_alloc_buf(rd_kafka_msgset_writer_t *msetw) {
* ProduceRequest header sizes
*/
switch (msetw->msetw_ApiVersion) {
case 9:
case 8:
case 7:
case 6:
Expand Down Expand Up @@ -353,9 +354,10 @@ static void rd_kafka_msgset_writer_alloc_buf(rd_kafka_msgset_writer_t *msetw) {
* Allocate iovecs to hold all headers and messages,
* and allocate auxilliery space for message headers, etc.
*/
msetw->msetw_rkbuf =
rd_kafka_buf_new_request(msetw->msetw_rkb, RD_KAFKAP_Produce,
msetw->msetw_msgcntmax / 2 + 10, bufsize);
msetw->msetw_rkbuf = rd_kafka_buf_new_flexver_request(
msetw->msetw_rkb, RD_KAFKAP_Produce,
msetw->msetw_msgcntmax / 2 + 10, bufsize,
msetw->msetw_ApiVersion >= 9);

rd_kafka_buf_ApiVersion_set(msetw->msetw_rkbuf, msetw->msetw_ApiVersion,
msetw->msetw_features);
Expand Down Expand Up @@ -442,19 +444,19 @@ rd_kafka_msgset_writer_write_Produce_header(rd_kafka_msgset_writer_t *msetw) {
rd_kafka_buf_write_i32(rkbuf, rkt->rkt_conf.request_timeout_ms);

/* TopicArrayCnt */
rd_kafka_buf_write_i32(rkbuf, 1);
rd_kafka_buf_write_arraycnt(rkbuf, 1);

/* Insert topic */
rd_kafka_buf_write_kstr(rkbuf, rkt->rkt_topic);

/* PartitionArrayCnt */
rd_kafka_buf_write_i32(rkbuf, 1);
rd_kafka_buf_write_arraycnt(rkbuf, 1);

/* Partition */
rd_kafka_buf_write_i32(rkbuf, msetw->msetw_rktp->rktp_partition);

/* MessageSetSize: Will be finalized later*/
msetw->msetw_of_MessageSetSize = rd_kafka_buf_write_i32(rkbuf, 0);
msetw->msetw_of_MessageSetSize = rd_kafka_buf_write_arraycnt_pos(rkbuf);

if (msetw->msetw_MsgVersion == 2) {
/* MessageSet v2 header */
Expand Down Expand Up @@ -1316,9 +1318,9 @@ rd_kafka_msgset_writer_finalize_MessageSet(rd_kafka_msgset_writer_t *msetw) {
RD_KAFKAP_MSGSET_V0_SIZE + msetw->msetw_messages_len;

/* Update MessageSetSize */
rd_kafka_buf_update_i32(msetw->msetw_rkbuf,
msetw->msetw_of_MessageSetSize,
(int32_t)msetw->msetw_MessageSetSize);
rd_kafka_buf_finalize_arraycnt(msetw->msetw_rkbuf,
msetw->msetw_of_MessageSetSize,
(int32_t)msetw->msetw_MessageSetSize);
}


Expand Down Expand Up @@ -1378,6 +1380,11 @@ rd_kafka_msgset_writer_finalize(rd_kafka_msgset_writer_t *msetw,
/* Finalize MessageSet header fields */
rd_kafka_msgset_writer_finalize_MessageSet(msetw);

/* Partition tags */
rd_kafka_buf_write_tags_empty(rkbuf);
/* Topics tags */
rd_kafka_buf_write_tags_empty(rkbuf);

/* Return final MessageSetSize */
*MessageSetSizep = msetw->msetw_MessageSetSize;

Expand Down
14 changes: 11 additions & 3 deletions src/rdkafka_request.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3294,7 +3294,7 @@ rd_kafka_handle_Produce_parse(rd_kafka_broker_t *rkb,
const int log_decode_errors = LOG_ERR;
int64_t log_start_offset = -1;

rd_kafka_buf_read_i32(rkbuf, &TopicArrayCnt);
rd_kafka_buf_read_arraycnt(rkbuf, &TopicArrayCnt, RD_KAFKAP_TOPICS_MAX);
if (TopicArrayCnt != 1)
goto err;

Expand All @@ -3303,7 +3303,8 @@ rd_kafka_handle_Produce_parse(rd_kafka_broker_t *rkb,
* and that it is the same that we requested.
* If not the broker is buggy. */
rd_kafka_buf_skip_str(rkbuf);
rd_kafka_buf_read_i32(rkbuf, &PartitionArrayCnt);
rd_kafka_buf_read_arraycnt(rkbuf, &PartitionArrayCnt,
RD_KAFKAP_PARTITIONS_MAX);

if (PartitionArrayCnt != 1)
goto err;
Expand All @@ -3325,7 +3326,7 @@ rd_kafka_handle_Produce_parse(rd_kafka_broker_t *rkb,
int i;
int32_t RecordErrorsCnt;
rd_kafkap_str_t ErrorMessage;
rd_kafka_buf_read_i32(rkbuf, &RecordErrorsCnt);
rd_kafka_buf_read_arraycnt(rkbuf, &RecordErrorsCnt, -1);
if (RecordErrorsCnt) {
result->record_errors = rd_calloc(
RecordErrorsCnt, sizeof(*result->record_errors));
Expand All @@ -3343,6 +3344,8 @@ rd_kafka_handle_Produce_parse(rd_kafka_broker_t *rkb,
result->record_errors[i].errstr =
RD_KAFKAP_STR_DUP(
&BatchIndexErrorMessage);
/* RecordError tags */
rd_kafka_buf_skip_tags(rkbuf);
}
}

Expand All @@ -3351,6 +3354,11 @@ rd_kafka_handle_Produce_parse(rd_kafka_broker_t *rkb,
result->errstr = RD_KAFKAP_STR_DUP(&ErrorMessage);
}

/* Partition tags */
rd_kafka_buf_skip_tags(rkbuf);
/* Topic tags */
rd_kafka_buf_skip_tags(rkbuf);

if (request->rkbuf_reqhdr.ApiVersion >= 1) {
int32_t Throttle_Time;
rd_kafka_buf_read_i32(rkbuf, &Throttle_Time);
Expand Down