Skip to content

Commit bbc6296

Browse files
kenneth-jiakenneth-jia
committed
Pick up librdkafka v1.5.2
1 parent 2be7bf8 commit bbc6296

File tree

5 files changed

+21
-11
lines changed

5 files changed

+21
-11
lines changed

‎.github/workflows/on_pull_request.yml‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ on: [push, pull_request]
44

55
env:
66
CPU_CORE_NUM : 2
7-
LIBRDKAFKA_VERSION: 1.5.0
7+
LIBRDKAFKA_VERSION: 1.5.2
88

99
jobs:
1010
build_gcc_debug:
@@ -57,6 +57,7 @@ jobs:
5757
cd $BUILD_SUB_DIR
5858
make -j${CPU_CORE_NUM} VERBOSE=1
5959
- name: Test
60+
timeout-minutes: 15
6061
run: |
6162
cd $BUILD_SUB_DIR
6263
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -121,6 +122,7 @@ jobs:
121122
cd $BUILD_SUB_DIR
122123
make -j${CPU_CORE_NUM} VERBOSE=1
123124
- name: Test
125+
timeout-minutes: 15
124126
run: |
125127
cd $BUILD_SUB_DIR
126128
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -185,6 +187,7 @@ jobs:
185187
cd $BUILD_SUB_DIR
186188
make -j${CPU_CORE_NUM} VERBOSE=1
187189
- name: Test
190+
timeout-minutes: 15
188191
run: |
189192
cd $BUILD_SUB_DIR
190193
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -249,6 +252,7 @@ jobs:
249252
cd $BUILD_SUB_DIR
250253
make -j${CPU_CORE_NUM} VERBOSE=1
251254
- name: Test
255+
timeout-minutes: 15
252256
run: |
253257
cd $BUILD_SUB_DIR
254258
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -312,6 +316,7 @@ jobs:
312316
cd $BUILD_SUB_DIR
313317
make -j${CPU_CORE_NUM} VERBOSE=1
314318
- name: Test
319+
timeout-minutes: 15
315320
run: |
316321
cd $BUILD_SUB_DIR
317322
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -375,6 +380,7 @@ jobs:
375380
cd $BUILD_SUB_DIR
376381
make -j${CPU_CORE_NUM} VERBOSE=1
377382
- name: Test
383+
timeout-minutes: 15
378384
run: |
379385
cd $BUILD_SUB_DIR
380386
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -439,6 +445,7 @@ jobs:
439445
cd $BUILD_SUB_DIR
440446
make -j${CPU_CORE_NUM} VERBOSE=1
441447
- name: Test
448+
timeout-minutes: 15
442449
run: |
443450
cd $BUILD_SUB_DIR
444451
export PATH=`pwd`/kafka_2.13-2.6.0/bin:$PATH
@@ -531,7 +538,7 @@ jobs:
531538
run: |
532539
cd $BUILD_SUB_DIR
533540
make VERBOSE=1
534-
- name: Test
541+
- name: Post Check
535542
run: |
536543
cd $BUILD_SUB_DIR
537544
ls -l doc/README.html

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ The `Modern C++ based Kafka API` is a layer of C++ wrapper based on [librdkafka]
66

77
Here is the [doxygen documentation for developers](doxygen/annotated.html).
88

9-
- At present, the `Modern C++ based Kafka API` is compatible with `librdkafka` v1.5.0.
9+
- At present, the `Modern C++ based Kafka API` is compatible with `librdkafka` v1.5.2.
1010

1111
## Why it's here
1212

‎doc/KafkaClientConfiguration.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,11 +28,11 @@ CLIENT_ID | Kafka Producer's ID string
2828
ACKS | How many partition replicas must receive the record before the producer can consider the write successful | 0, 1, all(-1) | all
2929
QUEUE_BUFFERING_MAX_MESSAGES | Maximum number of messages allowed on the producer queue | Integer[1, 10000000] | 100000
3030
QUEUE_BUFFERING_MAX_KBYTES | Maximum total message size sum allowed on the producer queue | Integer[1, INT_MAX] | 0x100000 (1GB)
31-
LINGER_MS | Delay in milliseconds to wait before constructing messages batches to transmit to brokers | Double[0, 900000.0] | 0 (KafkaSyncProducer);<br />0.5 (KafkaAsyncProducer)
31+
LINGER_MS | Delay in milliseconds to wait before constructing messages batches to transmit to brokers | Double[0, 900000.0] | 0 (KafkaSyncProducer);<br />5 (KafkaAsyncProducer)
3232
BATCH_NUM_MESSAGES | Maximum number of messages batched in one messageSet | Integer[1, 1000000] | 10000
3333
MESSAGE_MAX_BYTES | Maximum Kafka protocol request message (or batch) size | Integer[1000, 1000000000] | 1000000
3434
MESSAGE_TIMEOUT_MS | This value is enforced locally and limits the time a produced message waits for successful delivery | Integer[0, INT32_MAX] | 300000
35-
REQUEST_TIMEOUT_MS | This value is only enforced by the brokers and relies on `ACKS` being non-zero;<br /> It indicates how long the leading broker should wait for in-sync replicas to acknowledge the message. | Integer[1, 900000] | 5000
35+
REQUEST_TIMEOUT_MS | This value is only enforced by the brokers and relies on `ACKS` being non-zero;<br /> It indicates how long the leading broker should wait for in-sync replicas to acknowledge the message. | Integer[1, 900000] | 30000
3636
PARTITIONER | The default partitioner for a ProducerRecord (with no partition assigned).<br />Note: partitioners with postfix `_random` mean `ProducerRecord`s with empty key are randomly partitioned. | random,<br />consistent, consistent_random,<br />murmur2, murmur2_random,<br />fnv1a, fnv1a_random | murmur2_random
3737
MAX_IN_FLIGHT | Maximum number of in-flight requests per broker connection | Integer[1, 1000000] | 1000000 (while `enable.idempotence`=false);<br />5 (while `enable.idempotence`=true)
3838
ENABLE_IDEMPOTENCE | This feature ensures that messages are successfully sent exactly once and in the original order | true, false | false

‎tests/integration/TestKafkaProducer.cc‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,9 @@ TEST(KafkaSyncProducer, FailToSendMessagesWithAcksAll)
116116
KafkaTestUtility::CreateKafkaTopic(topic, numPartitions, replicationFactor);
117117

118118
// Properties for the producer
119-
const auto props = KafkaTestUtility::GetKafkaClientCommonConfig().put(ProducerConfig::ACKS, "all");
119+
const auto props = KafkaTestUtility::GetKafkaClientCommonConfig()
120+
.put(ProducerConfig::ACKS, "all")
121+
.put(ProducerConfig::MESSAGE_TIMEOUT_MS, "5000"); // To shorten the test
120122

121123
// Async-send producer
122124
KafkaSyncProducer producer(props);
@@ -131,8 +133,8 @@ TEST(KafkaSyncProducer, FailToSendMessagesWithAcksAll)
131133
{
132134
auto record = ProducerRecord(topic, Key(msg.first.c_str(), msg.first.size()), Value(msg.second.c_str(), msg.second.size()));
133135
std::cout << "[" << Utility::getCurrentTime() << "] ProducerRecord: " << record.toString() << std::endl;
134-
// "no in-sync replica" for the topic.
135-
EXPECT_KAFKA_THROW(producer.send(record), RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS);
136+
// Since "no in-sync replica" for the topic, it would keep trying
137+
EXPECT_KAFKA_THROW(producer.send(record), RD_KAFKA_RESP_ERR__MSG_TIMED_OUT);
136138
}
137139
}
138140

@@ -148,11 +150,12 @@ TEST(KafkaSyncProducer, InSyncBrokersAckTimeout)
148150
{
149151
const auto props = KafkaTestUtility::GetKafkaClientCommonConfig()
150152
.put(ProducerConfig::ACKS, "all")
153+
.put(ProducerConfig::MESSAGE_TIMEOUT_MS, "1000")
151154
.put(ProducerConfig::REQUEST_TIMEOUT_MS, "1"); // Here it's a short value, more likely to trigger the timeout
152155

153156
KafkaSyncProducer producer(props);
154157

155-
constexpr int MAX_RETRIES = 1000;
158+
constexpr int MAX_RETRIES = 100;
156159
for (int i = 0; i < MAX_RETRIES; ++i)
157160
{
158161
try
@@ -163,7 +166,7 @@ TEST(KafkaSyncProducer, InSyncBrokersAckTimeout)
163166
catch (const KafkaException& e)
164167
{
165168
std::cout << "[" << Utility::getCurrentTime() << "] Exception caught: " << e.what() << std::endl;
166-
EXPECT_EQ(RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT, e.error().value());
169+
EXPECT_EQ(RD_KAFKA_RESP_ERR__MSG_TIMED_OUT, e.error().value());
167170
break;
168171
}
169172
}

‎tests/unit/TestKafkaClientDefaultProperties.cc‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ TEST(KafkaClient, KafkaProducerDefaultProperties)
5555
{ Kafka::ProducerConfig::BATCH_SIZE, "1000000" },
5656
{ Kafka::ProducerConfig::MESSAGE_MAX_BYTES, "1000000" },
5757
{ Kafka::ProducerConfig::MESSAGE_TIMEOUT_MS, "300000" },
58-
{ Kafka::ProducerConfig::REQUEST_TIMEOUT_MS, "5000" },
58+
{ Kafka::ProducerConfig::REQUEST_TIMEOUT_MS, "30000" },
5959
{ Kafka::ProducerConfig::PARTITIONER, "murmur2_random" },
6060
{ Kafka::ProducerConfig::SECURITY_PROTOCOL, "plaintext" },
6161
{ Kafka::ProducerConfig::MAX_IN_FLIGHT, "1000000" },

0 commit comments

Comments
 (0)