Logstash failing to send to Kafka

Elastic Stack Logstash
1

Hi Team,

I am parsing Gihthub Log details from Logstash to Kafka and the logstash keeps throwing the below issue. After a certain time, I need to restart logstash to get rid of those warnings.

Once initiating the docker session, the logstash (Docker Containers) keeps throwing the below warn/err.

[2020-11-02T23:51:54,928][INFO ][logstash.outputs.kafka   ][httpd][da57d4766911e4fad530f7fa630c3790519cb207ca20be19ea4a6ff35424df6d] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>1, :failures=>1, :sleep=>0.1}
[2020-11-02T23:51:54,930][WARN ][logstash.outputs.kafka   ][httpd][da57d4766911e4fad530f7fa630c3790519cb207ca20be19ea4a6ff35424df6d] producer send failed {:exception=>Java::JavaUtilConcurrent::ExecutionException, :message=>"org.apache.kafka.common.errors.RecordTooLargeException: The message is 4369075 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration."}
[2020-11-02T23:51:54,930][INFO ][logstash.outputs.kafka   ][httpd][da57d4766911e4fad530f7fa630c3790519cb207ca20be19ea4a6ff35424df6d] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>1, :failures=>1, :sleep=>0.1}

In the kafka compose files, i have mentioned the necessary settings for the size but it doesn't helps.

Kafka Compose entry file

  kafka1:

    hostname: kafka-1
    ports:
      - "9092:9092"
    networks:
      - kafkanw
    volumes:
      - kafka1:/var/lib/kafka/data
    depends_on:
      - zookeeper1
      - zookeeper2
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181,zookeeper2:22181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.207.106.54:9092
      KAFKA_PRODUCER_MAX_REQUEST_SIZE: 15728640
      CONNECT_PRODUCER_MAX_REQUEST_SIZE: 15728640
      KAFKA_CONSUMER_MAX_REQUEST_SIZE: 15728640
      KAFKA_MESSAGE_MAX_BYTES: 9000000

Can you please help me understand if there is any other setting I need to perform so that to avoid the logstash warnings.

Regards
TJ

2

You should also configure Kafka producer besides Broker.

3

I have configured the Kafka Brokers with producer configurations. Please find the below dockerfile of kafka broker. :slightly_smiling_face:

  kafka1:
    image: tjkafka:1.0
    hostname: kafka-1
    ports:
      - "9092:9092"
    networks:
      - kafkanw
    volumes:
      - kafka1:/var/lib/kafka/data
    depends_on:
      - zookeeper1
      - zookeeper2
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper1:2181,zookeeper2:22181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://10.207.106.54:9092
      KAFKA_PRODUCER_MAX_REQUEST_SIZE: 15728640
      CONNECT_PRODUCER_MAX_REQUEST_SIZE: 15728640
      KAFKA_CONSUMER_MAX_REQUEST_SIZE: 15728640
      KAFKA_MESSAGE_MAX_BYTES: 9000000

As I mentioned in the initial draft, logstash keeps throwing the warnings/infos and after a period of time, logstash stops accepting any inputs(http).

I still see the console log as below

[2020-11-03T16:35:08,762][INFO ][logstash.outputs.kafka   ][httpd][da57d4766911e4fad530f7fa630c3790519cb207ca20be19ea4a6ff35424df6d] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>1, :failures=>1, :sleep=>0.1}
[2020-11-03T16:35:08,849][WARN ][logstash.outputs.kafka   ][httpd][da57d4766911e4fad530f7fa630c3790519cb207ca20be19ea4a6ff35424df6d] producer send failed {:exception=>Java::JavaUtilConcurrent::ExecutionException, :message=>"org.apache.kafka.common.errors.RecordTooLargeException: The message is 3378103 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration."}
[2020-11-03T16:35:08,849][INFO ][logstash.outputs.kafka   ][httpd][da57d4766911e4fad530f7fa630c3790519cb207ca20be19ea4a6ff35424df6d] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>1, :failures=>1, :sleep=>0.1}

Looking forward to hear from you.

4

I mean you should configure Kafka producer in Logstash Kafka module.

5
6

After adding the max_request_size, the warnings have faded away.

Adding the updated output logstash phase here for future/others reference. Is there any other data parameter which I need to capture in the logstash output filter to avoid any other issues in future?

output {
  stdout { codec => rubydebug { metadata => true } }
  kafka {
        codec => json
        topic_id => "githubevent"
        bootstrap_servers => "KAFKA:9092"
        max_request_size => 104857600
  }
}