BroadwayKafka.Producer (BroadwayKafka v0.6.1)

A Kafka connector for Broadway.

BroadwayKafka can subscribe as a consumer to one or more topics and process streams of records within the same consumer group. Communication is done through Kafka's Consumer API using the :brod client.

Options

• :hosts - Required. A list of host and port pairs, or one string of comma-separated HOST:PORT pairs, used to make the first connection to Kafka. For example: [localhost: 9092], [{"kafka-vm1", 9092}, {"kafka-vm2", 9092}], "kafka-vm1:9092,kafka-vm2:9092".

• :group_id - Required. A (non-empty) unique string that identifies the consumer group that the producer will join.

• :topics - Required. A list of topics that the producer will subscribe to.

• :receive_interval - The duration (in milliseconds) that the producer waits before making a request for more messages. The default value is 2000.

• :offset_commit_on_ack - Tells Broadway to send or not an offset commit request after each acknowledgement. Setting this value to false can increase performance since commit requests will respect the :offset_commit_interval_seconds option. However, setting long commit intervals might lead to a large number of duplicated records to be processed after a server restart or connection loss. If that's the case, make sure your logic is idempotent when consuming records to avoid inconsistencies. Also, bear in mind the the negative performance impact might be insignificant if you're using batchers since only one commit request will be performed per batch. The default value is true.

• :offset_reset_policy - The offset to use when Kafka has no saved offset or the saved offset has expired. Use :earliest, :latest, or {:timestamp, timestamp}, where timestamp is a non-negative time in milliseconds. The default value is :latest.

• :begin_offset - How consumers choose their first offset. When set to :assigned, the starting offset will be the one returned in the Kafka partition assignments (the latest committed offsets for the consumer group). When set to :reset, the starting offset will be dictated by the :offset_reset_policy option, either starting from the :earliest or the :latest offsets of the topic. The default value is :assigned.

• :shared_client - When false, it starts one :brod client per producer. When true, it starts a single shared :brod client across all producers (which may reduce memory/resource usage). May cause severe performance degradation, see "Shared Client Performance" for details. The default value is false.

• :group_config - Options passed to :brod's group coordinator. The default value is [].

• :fetch_config - The available options to configure how messages are fetched. Unless noted otherwise, they are internally passed to :brod.fetch/5. The default value is [].

• :client_config - Options passed to :brod.start_client/3 when it starts a client. The default value is [].

Group config

• :group_instance_id - A unique, non-empty string that identifies this consumer group member across restarts. This enables static group membership and requires :brod 4.6.3 or later. When Kafka fences a static member, BroadwayKafka stops that member instead of trying to take the ID back. Retrying can make old and new instances fence each other during a rolling deploy and cause repeated group rebalances. Available since v0.6.0.

• :offset_commit_interval_seconds - The time in seconds between two OffsetCommitRequest messages. The default value is 5.

• :rejoin_delay_seconds - The delay in seconds before rejoining the group. The default value is 1.

• :session_timeout_seconds - The time in seconds that the group coordinator waits before it considers a member "down" (if no heartbeat or any kind of request is received). A group member may also consider the coordinator "down" if it receives no heartbeat response within this time. The default value is 30.

• :heartbeat_rate_seconds - The time in seconds between heartbeats ("ping" requests) sent to the group coordinator. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than :session_timeout_seconds, typically equal to or lower than ⅓ of that value. It can be adjusted even lower to control the expected time for normal rebalances. The default value is 5.

• :rebalance_timeout_seconds - The time in seconds that each worker has to join the group after a rebalance starts. If the timeout is exceeded, then the worker will be removed from the group, which will cause offset commit failures. The default value is 30.

Fetch config

• :min_bytes - The minimum amount of data to be fetched from the server. If not enough data is available the request will wait for that much data to accumulate before answering. Setting this value greater than 1 can improve server throughput a bit at the cost of additional latency. The default value is 1.

• :max_bytes - The most data to fetch from one partition at a time. A larger value may improve throughput at the cost of more memory use. The default value is 1048576.

• :max_wait_time - The most time (in millisecond) that the broker may wait for :min_bytes of data. The default value is 1000.

• :max_fetch_retries - How many times a failed fetch is retried in place when Kafka returns a retriable error, before the producer raises as it does for any other error. Set to 0 to disable retries. BroadwayKafka uses this option itself and does not pass it to :brod.fetch/5. The default value is 3.

• :fetch_retry_backoff_ms - The time, in milliseconds, to wait before the first retry of a failed fetch (see :max_fetch_retries). The wait time doubles on each later retry. BroadwayKafka uses this option itself and does not pass it to :brod.fetch/5. The default value is 500.

Client config

• :client_id_prefix - A string added to the client ID that BroadwayKafka builds for :brod.

• :sasl - A tuple of "mechanism" (:plain, :scram_sha_256, or :scram_sha_512), username, and password. See :brod's Authentication Support documentation for more information. You may also pass {:callback, module, options} for a SASL plug-in. Set this to :undefined to disable SASL.

• :ssl - A boolean or a keyword list of SSL/TLS client options. See the tls_client_option documentation for more information.

• :connect_timeout - The time (in milliseconds) allowed for a connection to Kafka. Default is what :brod defaults to (5s at the time of writing).

• :request_timeout - The time (in milliseconds) allowed for a response from Kafka. It must be at least 1_000. Default is to use :brod's default timeout which is currently 4 minutes (240_000).

• :query_api_versions - Whether to ask Kafka which API versions it supports when a connection starts. Set this to false for Kafka versions before 0.10.

• :extra_sock_opts - Extra gen_tcp socket options. More info. Set to [:inet6] if your Kafka broker uses IPv6.

• :allow_topic_auto_creation - Whether :brod may send metadata requests that can create a topic when the broker allows automatic topic creation.

Example

Broadway.start_link(MyBroadway,
  name: MyBroadway,
  producer: [
    module: {BroadwayKafka.Producer, [
      hosts: [localhost: 9092],
      group_id: "group_1",
      topics: ["test"],
    ]},
    concurrency: 1
  ],
  processors: [
    default: [
      concurrency: 10
    ]
  ]
)

Concurrency and partitioning

The concurrency model provided by Kafka is based on partitioning, i.e., the more partitions you have, the more concurrency you get. However, in order to take advantage of this model you need to set up the :concurrency options for your processors and batchers accordingly. Having less concurrency than topic/partitions assigned will result in individual processors handling more than one partition, decreasing the overall level of concurrency. Therefore, if you want to always be able to process messages at maximum concurrency (assuming you have enough resources to do it), you should increase the concurrency up front to make sure you have enough processors to handle the extra messages received from new partitions assigned.

Note: Even if you don't plan to add more partitions to a Kafka topic, your pipeline can still receive more assignments than planned. For instance, if another consumer crashes, the server will reassign all its topic/partition to other available consumers, including any Broadway producer subscribed to the same topic.

Handling failed messages

BroadwayKafka never stops the flow of the stream, i.e. it will always ack the messages even when they fail. Unlike queue-based connectors, where you can mark a single message as failed. In Kafka that's not possible due to its single offset per topic/partition ack strategy. If you want to reprocess failed messages, you need to roll your own strategy. A possible way to do that is to implement Broadway.handle_failed/2 and send failed messages to a separated stream or queue for later processing.

Message metadata

When producing messages, the following information will be passed to Broadway.Message's metadata.

• topic - The topic the message was published.

• partition - The topic partition.

• offset - The offset assigned to the message inside the partition.

• key - The partition key.

• ts - A timestamp associated with the message.

• headers - The headers of the message.

Telemetry

This producer emits a few Telemetry events which are listed below.

• [:broadway_kafka, :assignments_revoked, :start | :stop | :exception] spans - these events are emitted in "span style" when receiving assignments revoked call from consumer group coordinator See :telemetry.span/3.

• [:broadway_kafka, :fenced_instance_id] - emitted after a producer stops consuming because Kafka fenced its static group member. The measurement is :system_time. The metadata includes :producer, :client_id, :group_id, and :group_instance_id.

Shared Client Performance

Enabling shared client may drastically decrease performance. Since connection is handled by a single process, producers may block each other waiting for the client response.

This is more likely to be an issue if the producers on your pipeline are fetching message from multiple topics and specially if there are very low traffic topics, which may block on batch wait times.

To mitigate this, you can split your topics between multiple pipelines, but notice that this will increase the resource usage as well. By creating one new client/connection for each pipeline, you effectively diminishing the shared_client resource usage gains. So make sure to measure if you enable this option.