Additional parameters for Kafka input

[ppavlov39]

Hello!
We were experiencing a strange stop in processing messages from Kafka (using kafka input). There are 12 partitions in the topic and we launch 12 pods to read them, but almost immediately most of the pods stop processing anything.
We use an http client as output. Both Kafka input and HTTP output use batch processing (batch size is 2000).
The complete configuration of these components is shown below.
We have a hypothesis that changing some timeouts in the Kafka consumer (sarama) can help us.
Is there a way to configure settings that Kafka input doesn't have?

input:
kafka:
      addresses: <>
      topics: <>
      consumer_group: <>
      client_id: <>
      checkpoint_limit: 1
      target_version": "2.4.0"
      max_processing_period: "300ms"
      group.heartbeat_interval: "10s"
      batching: 
        count: 2000
        period: 120s

output:
http_client: 
          url: <>
          verb: "POST"
          rate_limit: ""
          timeout: "120s"
          headers:
            Content-Type": "application/json"
            Content-Encoding": "gzip"
            ...
          max_in_flight: 1
          batching: 
            count: 2000
            period: 120s
            processor": 
              archive:
                  format: lines
              compress:
                  algorithm: gzip
                  level: 5

[mfamador]

You probably need to increase checkpoint_limit to something like 2000.

https://www.benthos.dev/docs/components/inputs/kafka/#checkpoint_limit

[ppavlov39]

We deliberately set the checkpoint limit to 1 because it is important to process messages in a strict order.
And I can confirm that setting the parameter to a higher value will solve the problem.

[mihaitodor]

In that case, batching won't work because you're telling the input that only one message / topic can be processed at any given time, but then you configured input batching at the input level (which applies per topic), so Benthos isn't able to consume enough messages to fill those batches. You also have output batching, but that ignores the partitions and is redundant in this case, because it has the same size as the input batching (note that if you set a smaller size here, it will be ignored, since Benthos only shrinks batches using the split processor). If you want to archive & compress your batches, you can add processors at the output level (see here) or under input.kafka.batching.processors. Since you also have max_in_flight: 1, that tells Benthos to only allow one batch at a time to be sent to the output. I assume you care about message order per partition, so then that's ensured by having input batching (because it's per partition) and checkpoint_limit set to some value that's bigger than the number of messages you'll have in your batch. If you leave max_in_flight: 1, that guarantees that you can't have 2 batches from the same partition in flight at the same time, so they won't get jumbled up.

LE: Oops, I jumped the gun there without checking and assumed that checkpoint_limit: 1 doesn't work with batching. My bad!

[ppavlov39]

Thanks for the answer!

I have some doubts about this. The sarama_kafka module states that checkoint_limit must be equal to 1 to ensure strict ordering of message processing. And my understanding of this is that setting a value higher than 1 may cause messages to be processed out of order. So we might have the wrong order of messages into the batches.
We also see a problem with a topic that has millions of messages, so benthos should be able to fill multiple batches for any partition. Any way in input batch policy we set a period (120s) and Benthos must flush batches at this interval, even with one message in it.

Could you explain whats wrong with output.http_client.batching.processor.{archive|compress} and how checkpoint_limit` works, please?

[mihaitodor]

Hey @ppavlov39, my bad, it's fine to have batching with checkpoint_limit: 1, sorry for the incorrect info. I should've checked first. I edited my answer.

We also see a problem with a topic that has millions of messages, so benthos should be able to fill multiple batches for any partition.

Only if you set max_in_flight to be bigger than one, but then I think it will break the ordering because those batches will be sent to the output in parallel.

Could you explain whats wrong with output.http_client.batching.processor.{archive|compress}

It's not wrong, but I think you should use output.processors to avoid having messages pass through another batcher which is doing redundant bookkeeping.

Regarding your original question, if you're familiar with Go, then I think the best way to confirm your hunch would be to clone the code, make the timeout adjustments that you believe are needed over here and then run make to build your own Benthos binary (it will pop up under ./target/bin/) that you can test. If you can confirm that, indeed, adjusting those settings does help, then it should be straightforward to expose them as extra config parameters. PRs are welcome, but if you need help with the plumbing, I'm happy to do that if you let us know which parameters have to be exposed.

[ppavlov39]

Hey @ppavlov39, my bad, it's fine to have batching with checkpoint_limit: 1, sorry for the incorrect info. I should've checked first. I edited my answer.

Thanks)

We also see a problem with a topic that has millions of messages, so benthos should be able to fill multiple batches for any partition.

Only if you set max_in_flight to be bigger than one, but then I think it will break the ordering because those batches will be sent to the output in parallel.

I've meant that Benthos shouldn't have any problems with filling up the batches even with max_in_flight=1 since all topic partitions have many messages.

Regarding your original question, if you're familiar with Go, then I think the best way to confirm your hunch would be to clone the code, make the timeout adjustments that you believe are needed over here and then run make to build your own Benthos binary (it will pop up under ./target/bin/) that you can test. If you can confirm that, indeed, adjusting those settings does help, then it should be straightforward to expose them as extra config parameters. PRs are welcome, but if you need help with the plumbing, I'm happy to do that if you let us know which parameters have to be exposed.

Thanks for your reply. But unfortunately, I can't reproduce the problem now... :( When this problem occurs again, I will try changing Sarama settings to see if that helps or not.