Kafka rebalancing never finishes or fails when Broadway Producer message processing takes too long

Viewed 280

Running 6 instances of my Broadway Producer application. My processing of messages may take 5 min plus. When this happens Kafka coordinator never finishes rebalancing. How can I increase the proper Kafka consumer group option max.poll.interval.ms in Brod via Broadway group options in order to increase the time my app has to process a message. Running Kafka 1.0.1 and the following

── brod ~> 3.14.0 (Hex package) *override*
│   ├── kafka_protocol 2.3.6 (Hex package)
│   │   ├── crc32cer 0.1.4 (Hex package)
│   │   └── snappyer 1.2.5 (Hex package)
│   └── supervisor3 1.1.11 (Hex package)
├── broadway_kafka ~> 0.1.0 (Hex package)
│   ├── broadway ~> 0.6.0 (Hex package)
│   │   ├── gen_stage ~> 1.0 (Hex package)
│   │   └── telemetry ~> 0.4.0 (Hex package)
│   └── brod ~> 3.10 (Hex package)

My Broadway config is like this...

  @moduledoc """
  """
  use Broadway
  alias Broadway.Message

  require Logger

  def start_link(_opts) do

    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwayKafka.Producer, [
          hosts: [{"aws-test-kafka01", 9092}],
          group_id: "my-kafka-consumer-group",
          topics: ["my-broadway-topic"],
          offset_reset_policy: :earliest,
          fetch_config: [
            max_bytes: 2048
          ],
          group_config: [
            rejoin_delay_seconds: 10,
            offset_commit_interval_seconds: 5,
            session_timeout_seconds: 10
          ],
        ]},
        concurrency: 12
      ],
      processors: [
        default: [
          concurrency: 2
        ]
      ]
    )
  end

  @impl true
  def handle_message(processor, message, context) do

    # Some 5 min + processing

    message
  end

  @impl true
  def handle_failed(messages, _context) do
    Logger.info("Failed messages #{inspect(messages)}")
    messages
  end

end

The topic I am consuming has 12 partitions

0 Answers
Related