retry on franz consumer group session errors

[samirketema]

Description

Implements retry logic for the kafka consumer on group join and leader errors.

Quick checks:

• I have followed the Code Guidelines.
• There is no other pull request for the same update/change.
• I have written unit tests.
• I have made sure that the PR is of reasonable size and can be easily reviewed.

Manual Test

Steps to manually test:

1. Spin up a kafka cluster using this docker compose:

version: '2'
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka1:
    image: wurstmeister/kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://:9092
      KAFKA_ADVERTISED_LISTENERS: INSIDE://<your-local-ip>:9092
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
    depends_on:
      - zookeeper
  kafka2:
    image: wurstmeister/kafka
    ports:
      - "9093:9093"
    environment:
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://:9093
      KAFKA_ADVERTISED_LISTENERS: INSIDE://<your-local-ip>:9093
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
    depends_on:
      - zookeeper
  kafka3:
    image: wurstmeister/kafka
    ports:
      - "9094:9094"
    environment:
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT
      KAFKA_LISTENERS: INSIDE://:9094
      KAFKA_ADVERTISED_LISTENERS: INSIDE://<your-local-ip>:9094
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
    depends_on:
      - zookeeper

2. Run the following pipelines against the kafka cluster:

version: "2.2"
pipelines:
  - id: gen-kafka
    status: running
    connectors:
      - id: gen.in
        type: source
        plugin: builtin:generator
        settings:
          format.type: "structured"
          format.options.id: "int"
          format.options.name: "string"
          format.options.company: "string"
          format.options.trial: "bool"
          rate: 0.5
          recordCount: "1000"
      - id: kafka.out
        type: destination
        plugin: standalone:kafka
        settings:
          batchBytes: "10485760"
          sdk.batch.delay: 1s
          sdk.batch.size: "50000"
          servers: <your-local-ip>:9092,<your-local-ip>:9093,<your-local-ip>:9094
          topic: test
  - id: kafka-log
    status: running
    connectors:  
      - id: kafka.in
        type: source
        plugin: standalone:kafka
        settings:
          readFromBeginning: "true"
          servers: <your-local-ip>:9092,<your-local-ip>:9093,<your-local-ip>:9094
          topic: test
        processors:
          - id: unwrapopencdc
            type: builtin:unwrap.opencdc
            plugin: ""
            condition: ""
            settings:
              field: .Payload.After
      - id: log.out
        type: destination
        plugin: builtin:log

3. Simulate a node going down by pausing or stopping one of the kafka node containers:

Before

Note that the pipeline errors out and stops, preventing a recovery from the issue:

2024-07-10T20:49:31+00:00 ERR pipeline stopped error="node kafka-log:kafka.in stopped with error: source stream was stopped unexpectedly: error reading from source: read plugin error: failed getting a record: unable to join group session: unable to dial: dial tcp 127.0.0.1:9092: connect: connection refused" component=pipeline.Service pipeline_id=kafka-log stack=[{"file":"/Users/samir/go/src/conduit/pkg/pipeline/lifecycle.go","func":"github.com/conduitio/conduit/pkg/pipeline.(*Service).runPipeline.func2","line":582},{"file":"/Users/samir/go/src/conduit/pkg/pipeline/stream/source.go","func":"github.com/conduitio/conduit/pkg/pipeline/stream.(*SourceNode).Run","line":140},{"file":"/Users/samir/go/src/conduit/pkg/pipeline/stream/source.go","func":"github.com/conduitio/conduit/pkg/pipeline/stream.(*SourceNode).Run.func1","line":87}]

After

Note that the errors are logged, but are continually retried with ErrBackoffRetry. This way, it's not silent, but we still can show the error.

After the broker comes back up, we can see that the connector continues processing records.

2024-07-10T20:08:17+00:00 ERR group session error, retrying: unable to join group session: unable to dial: dial tcp 127.0.0.1:9093: connect: connection refused component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka
2024-07-10T20:08:17+00:00 ERR join and sync loop errored component=plugin.standalone connector_id=kafka-log:kafka.in group=backoff plugin_name=conduit-connector-kafka
2024-07-10T20:08:17+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:17+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=gen-kafka:kafka.out plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:17+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:18+00:00 WRN unable to open connection to broker addr=err component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka
2024-07-10T20:08:18+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=gen-kafka:kafka.out plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:18+00:00 WRN produce partition load error, bumping error count on first stored batch broker=will_fail component=plugin.standalone connector_id=gen-kafka:kafka.out plugin_name=conduit-connector-kafka
2024-07-10T20:08:19+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:19+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:20+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:21+00:00 WRN produce partition load error, bumping error count on first stored batch broker=will_fail component=plugin.standalone connector_id=gen-kafka:kafka.out plugin_name=conduit-connector-kafka
2024-07-10T20:08:21+00:00 WRN read from broker errored, killing connection after 0 successful responses (is SASL missing?) component=plugin.standalone connector_id=kafka-log:kafka.in plugin_name=conduit-connector-kafka req=err
2024-07-10T20:08:26+00:00 INF component=plugin connector_id=kafka-log:log.out plugin_name=builtin:log plugin_type=destination record={"key":"c2FjY2hhcmlmeQ==","metadata":{"conduit.source.connector.id":"gen-kafka:gen.in","opencdc.readAt":"1720667249647356000"},"operation":"create","payload":{"after":{"company":"redipper","id":1592616014876622300,"name":"oenanthaldehyde","trial":false},"before":null},"position":"eyJHcm91cElEIjoiOWIyNzFkN2UtYTBmNC00MzlhLTk4YWMtN2YyMjk2NjIxOTdhIiwiVG9waWMiOiJzYW1pci10ZXN0LTIiLCJQYXJ0aXRpb24iOjAsIk9mZnNldCI6MH0="}

[samirketema]

@lovromazgon ok, feedback has been addressed - thank you 🙇🏾

edit - also, I re-tested with these changes, and still works as expected 🙏🏾

[lovromazgon]

👏