Getting WARNING logs related to kafka sink

[ShahroZafar]

I am getting these logs for kafka sink which doesn't make sense. Does anybody know about them?
I am on vector version 0.38.0

2024-05-17T06:16:51.858517Z  WARN librdkafka: librdkafka: CONFWARN [thrd:app]: Configuration property queue.buffering.max.ms is a producer property and will be ignored by this consumer instance
2024-05-17T06:16:51.858546Z  WARN librdkafka: librdkafka: CONFWARN [thrd:app]: Configuration property batch.size is a producer property and will be ignored by this consumer instance
2024-05-17T06:16:51.858550Z  WARN librdkafka: librdkafka: CONFWARN [thrd:app]: Configuration property request.required.acks is a producer property and will be ignored by this consumer instance
2024-05-17T06:16:51.858552Z  WARN librdkafka: librdkafka: CONFWARN [thrd:app]: Configuration property message.timeout.ms is a producer property and will be ignored by this consumer instance

[jszwedko]

Can you share your configuration?

[ShahroZafar]

I wonder if these warning logs are just to the extent of logs or librdkafka_options doesn't take effect at all

[jszwedko]

Oh I see what's happening here. The "healthcheck" is run by a Kafka consumer instance, but the rdkafka options are also applied ot it (in addition to the producer instance for sending messages). That is definitely confusing. I think ideally the logs would make that clearer.

[ShahroZafar]

So if I understand correctly, 2 instances are created consumer and producers?

consumer -> for healthchecks, librdkafka_options ignored
producer -> pushing logs to kafka and librdkafka_options are indeed applied?

[jszwedko]

Well, for healthchecks, only options that only apply to consumers are ignored. There are some options that can apply to both consumers and producers that will still apply to healthchecks. For the producer, you are correct.

[belltoy]

I have the same question about this warning.

After trace back the commit history, I wonder why the original commit for healthcheck of Kafka sink use a consumer instead of a lower level client back in 2019 (b8de832). I guess the reason is the high level producer (Future and Threaded) in rdkafka doesn't provide fetch_metadata API.

Another question is why the KafkaRole introduced in the commit 69c398a. Was it to distinguish the Consumer usage in the healthcheck function? If it was, I think the code of healthcheck in the Kafka sink can be simplified.

[jszwedko]

After trace back the commit history, I wonder why the original commit for healthcheck of Kafka sink use a consumer instead of a lower level client back in 2019 (b8de832). I guess the reason is the high level producer (Future and Threaded) in rdkafka doesn't provide fetch_metadata API.

It would seem sensible to me to just fetch the metadata in the healthcheck to verify connectivity. It is actually available via https://docs.rs/rdkafka/latest/rdkafka/client/struct.Client.html#method.fetch_metadata. The client is available on FutureProducer: https://docs.rs/rdkafka/latest/rdkafka/producer/trait.Producer.html#tymethod.client. I'd support a PR for that.

Another question is why the KafkaRole introduced in the commit 69c398a. Was it to distinguish the Consumer usage in the healthcheck function? If it was, I think the code of healthcheck in the Kafka sink can be simplified.

Agreed, I think it could be simplified if we avoid using a Kafka consumer in the sink.

Thanks for the thoughts @belltoy ! Let me know if you'd be interested in contributing these proposed changes.

[belltoy]

I'd like to commit a PR (#21129) for that.