help configuring kafka sink

[thisis2394]

Hi Everyone,

I tried posting same question in discord but did not get any response. Trying to repost it here in case anyone can help me out.

i am trying to push my webserver logs to kafka(MSK) but for some reason it seems to fail continuously and it is unclear to me what needs to done to fix this. Earlier i used to push from filebeat it worked fine can somone help me out on this please

my setup details

vector version: vector 0.44.0 (x86_64-unknown-linux-gnu 3cdc7c3 2025-01-13 21:26:04.735691656)
MSK/kafka(unauthenticated) : Apache Kafka: 3.5.1

vector conf:

api:
  enabled: true

sources:
  test_source:
    type: "file"
    include:
      - "/var/log/access/*.ind"

sinks:
  test_kafka_sink:
    type: kafka
    inputs:
      - test_source
    bootstrap_servers: "b-3.dummy.amazonaws.com:9092,b-2.dummy.amazonaws.com:9092,b-1.dummy.amazonaws.com:9092"
    topic: "ondemand-accesslog"
    encoding:
      codec: json

vector error:

2025-01-16T18:15:36.103476Z ERROR sink{component_kind="sink" component_id=test_kafka_sink component_type=kafka}:request{request_id=16}: vector_common::internal_event::service: Service call failed. No retries or retries exhausted. error=Some(KafkaError (Message production error: MessageTimedOut (Local: Message timed out))) request_id=16 error_type="request_failed" stage="sending" internal_log_rate_limit=true
2025-01-16T18:15:36.103571Z ERROR sink{component_kind="sink" component_id=test_kafka_sink component_type=kafka}:request{request_id=16}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=1 reason="Service call failed. No retries or retries exhausted."
 internal_log_rate_limit=true
2025-01-16T18:15:36.103591Z ERROR sink{component_kind="sink" component_id=test_kafka_sink component_type=kafka}:request{request_id=9}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] is being suppressed to avoid flooding.
2025-01-16T18:15:36.103603Z ERROR sink{component_kind="sink" component_id=test_kafka_sink component_type=kafka}:request{request_id=9}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] is being suppressed to avoid flooding.

[pront]

Hi @thisis2394, the error you're seeing indicates that Vector is having trouble sending messages to Kafka due to a timeout. The error MessageTimedOut (Local: Message timed out) suggests that the Kafka producer is unable to successfully send messages within the expected time frame.

Can you ensure that your brokers are reachable and that the topic exists?

You can try setting https://vector.dev/docs/reference/configuration/sinks/kafka/#message_timeout_ms to a higher value.

[thisis2394]

Hi @pront,

Thanks for taking time to respond to my question. This issue was due to misconfiguration in my Kafka topic. I tried producing some sample messages using kafka-console-producer and it highlighted what was happening once i fixed the issue in Kafka the log push seems to be working now.

Was wondering why vector was not highlighting the same error. I had enabled VECTOR_LOG=debug earlier while debugging but i had not noticed any such errors.

This was the error which Kafka CLI gave me:

org.apache.kafka.common.errors.NotEnoughReplicasException: Messages are rejected since there are fewer in-sync replicas than required.

[pront]

Glad that it works now 🎉

Was wondering why vector was not highlighting the same error. I had enabled VECTOR_LOG=debug earlier while debugging but i had not noticed any such errors.

If you can think of betters diagnostics, happy to hear more. And PRs are always very welcome. Closing this thread for now but feel free to re-open for any reason.