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