Skip to content

in_kafka: boost throughput #9625

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

Closed
wants to merge 3 commits into from
Closed

in_kafka: boost throughput #9625

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8f7c509
in_kafka: add support to switch to auto-commit
coreidcc Nov 20, 2024
6357338
in_kafka: increase the poll-timeout if we run in own own thread
coreidcc Nov 20, 2024
f94cf72
in_kafka: fix type in help text
coreidcc Nov 22, 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
29 changes: 26 additions & 3 deletions plugins/in_kafka/in_kafka.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ static int in_kafka_collect(struct flb_input_instance *ins,
ret = FLB_EVENT_ENCODER_SUCCESS;

while (ret == FLB_EVENT_ENCODER_SUCCESS) {
rkm = rd_kafka_consumer_poll(ctx->kafka.rk, 1);
rkm = rd_kafka_consumer_poll(ctx->kafka.rk, ctx->poll_timeount_ms);

if (!rkm) {
break;
Expand All @@ -180,8 +180,11 @@ static int in_kafka_collect(struct flb_input_instance *ins,

rd_kafka_message_destroy(rkm);

/* TO-DO: commit the record based on `ret` */
rd_kafka_commit(ctx->kafka.rk, NULL, 0);

if(!ctx->enable_auto_commit) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Current behaviour is to have auto-commit false but the default below is now true which means we are not maintaining the old behaviour by default.

/* TO-DO: commit the record based on `ret` */
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we need to sort all TODOs, we can't just leave them in there. I know it was in the original but we should attempt to sort or add more info.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well i din't introduce the TO-DO. It existed before. It just went into the if statement. Unfortunately github shows to little context so you can't see it.

rd_kafka_commit(ctx->kafka.rk, NULL, 0);
}

/* Break from the loop when reaching the limit of polling if available */
if (ctx->polling_threshold != FLB_IN_KAFKA_UNLIMITED &&
Expand Down Expand Up @@ -243,6 +246,21 @@ static int in_kafka_init(struct flb_input_instance *ins,
goto init_error;
}

/* Set the kafka poll timeout dependend on wether we run in our own
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Slightly modified the comment here.

    /* Set the Kafka poll timeout depending on whether we run in our own 
     * or in the main event thread. 
     * a) Run in main event thread: 
     *    -> Minimize the delay we might create
     * b) Run in our own thread: 
     *    -> Optimize for throughput and rely on 'fetch.wait.max.ms' 
     *       which is set to 500ms by default. Let's set it to
     *       twice so that increasing fetch.wait.max.ms still
     *       has an effect.
     */

* or in the main event thread.
* a) run in main event thread:
* -> minimize the delay we might create
* b) run in our own thread:
* -> optimize for throuput and relay on 'fetch.wait.max.ms'
* which is set to 500 by default default. lets set it to
* twice that so that increasing fetch.wait.max.ms still
* has an effect.
*/
ctx->poll_timeount_ms = 1;
if(ins->is_threaded) {
ctx->poll_timeount_ms = 1000;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wondering if poll_timeout_ms can be exposed as a config option? Given that rdkafka.fetch.wait.max.ms can be modified to fit the needs.

}

if (ctx->buffer_max_size > 0) {
ctx->polling_threshold = ctx->buffer_max_size;

Expand Down Expand Up @@ -428,6 +446,11 @@ static struct flb_config_map config_map[] = {
0, FLB_TRUE, offsetof(struct flb_in_kafka_config, buffer_max_size),
"Set the maximum size of chunk"
},
{
FLB_CONFIG_MAP_BOOL, "enable_auto_commit", FLB_IN_KAFKA_ENABLE_AUTO_COMMIT,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think so we need to this config option as this is included in the rdkafka.enable.auto.commit

0, FLB_TRUE, offsetof(struct flb_in_kafka_config, enable_auto_commit),
"Rely on kafka auto-commit and commit messages in batches"
},
/* EOF */
{0}
};
Expand Down
3 changes: 3 additions & 0 deletions plugins/in_kafka/in_kafka.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
#define FLB_IN_KAFKA_DEFAULT_FORMAT "none"
#define FLB_IN_KAFKA_UNLIMITED (size_t)-1
#define FLB_IN_KAFKA_BUFFER_MAX_SIZE "4M"
#define FLB_IN_KAFKA_ENABLE_AUTO_COMMIT "false"

enum {
FLB_IN_KAFKA_FORMAT_NONE,
Expand All @@ -48,6 +49,8 @@ struct flb_in_kafka_config {
int coll_fd;
size_t buffer_max_size; /* Maximum size of chunk allocation */
size_t polling_threshold;
bool enable_auto_commit;
int poll_timeount_ms;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

:nit: please fix the typo poll_timeount_ms -> poll_timeout_ms

};

#endif
Loading