[receiver/kafka] backoff in case of next consumer error

[yiquanzhou]

Description

Currently if the next consumer returns an error, kafka receiver simply returns the error and will consume the next message without any backpressure. This behavior is not optimal in case of some errors. For example the memory limiter could return data refused error when the memory usage of the collector is too high. Keeping consuming and sending messages to the memory limiter could further increase the memory usage and cause OutOfMemory error in the collector.

This PR provides an optional error backoff config which allows to wait before consuming the next message in case of errors that require backoff.

Testing

• Added a test case to TestXXXConsumerGroupHandler_error_nextConsumer tests with an error that requires backoff.

Documentation

• Added the configuration for error_backoff

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: yiquanzhou / name: Yiquan Zhou (2c0d768, 4411450, 704560b, 241c3b4, ca00ad8, 47a699c, 78a77fe, 6009470, 14c5863, 4ade9fa, 4019f7f)

[atoulme]

should messages be marked if you're going to retry them?

[yiquanzhou]

Currently if the next consumer returns an error, the kafka receiver simply drops the message and returns the error. At the beginning I intended to keep this behavior and only implement backoff delays without retrying the message. But then I went through the code again and I think we could implement the retry without introducing too much complexity.

I've updated the code and added some comments to explain. Let me know if that makes sense.

I've updated the unit test but I wonder how I could test this change in a integration or e2e test, in particular the retry logic if the offset is correctly reset and the failed message is consumed again in the next loop. Do you have any suggestion?

[yiquanzhou]

Hi @atoulme , could you please take another look at the changes that I added for the retry logic ? Also I'd appreciate some guidance on the testing.

[yiquanzhou]

@atoulme could you please check out the changes and my comments? I'd like to move this PR forward

[yiquanzhou]

I've tested the change by running the collector locally with the following config:

  kafka:
    encoding: text_utf-8 // the payload are decoded as text and inserted as the body of a log record
    error_backoff:
      enabled: true
      initial_interval: 5s
      max_interval: 1m
      multiplier: 1.5
      randomization_factor: 0
      max_elapsed_time: 1m
...
  memory_limiter:
    check_interval: 1s
    limit_mib: 30 // set a low memory limit to trigger the "data refused due to high memory usage" error 
    spike_limit_mib: 0

...
  pipelines:
    logs:
      receivers:
      - otlp
      - kafka
      processors:
      - memory_limiter
      - batch
      exporters:
      - debug

I also ran a kafka broker locally and use kafka-producer-perf-test tool to send messages to otlp_logs topic

$ ./kafka-producer-perf-test \
  --topic otlp_logs \
  --throughput 100000 \
  --num-records 1000000 \
  --record-size 1024 \
  --producer-props bootstrap.servers=localhost:9092

By adding some temporary logging, I can validate that when memory limit processor is returning data refused due to high memory usage error, the backoff is triggered and the same message will be retried after the backoff interval. Once the max_elapsed_time is exceeded, the backoff and retry is skipped and the receiver will continue consuming next messages.

[yiquanzhou]

I also added an entry to the change log.
@atoulme @MovieStoreGuy @pavolloffay Could any of you please give a final review of the PR?

[github-actions]

This PR was marked stale due to lack of activity. It will be closed in 14 days.

[MovieStoreGuy]

Does this method return any form of error, and what is the purpose of the empty string here?

[yiquanzhou]

No this method does not return error.
https://github.com/IBM/sarama/blob/main/consumer_group.go#L788

The empty string is the metadata parameter. We are also passing an empty string as metadata when calling session.MarkMessage at
https://github.com/dash0hq/opentelemetry-collector-contrib/blob/47a699cb7f71d6bd1f1ea50e237596e4673e92a1/receiver/kafkareceiver/kafka_receiver.go#L609