group_subscriber

Types

Builder

opaque </>

A builder for creating and configuring a Kafka consumer group subscriber.

pub opaque type Builder(callback_init_state)

CallbackReturn

</>

Return type for group subscriber callbacks.

pub type CallbackReturn

Config

</>

Configuration options for Kafka consumer groups.

pub type Config {
  SessionTimeoutSeconds(Int)
  RebalanceTimeoutSeconds(Int)
  HeartbeatRateSeconds(Int)
  MaxRejoinAttempts(Int)
  RejoinDelaySeconds(Int)
  OffsetCommitIntervalSeconds(Int)
  OffsetRetentionSeconds(Int)
}

Constructors

```
SessionTimeoutSeconds(Int)
```
Time in seconds for the group coordinator broker to consider a member ‘down’ if no heartbeat or any kind of requests received from a broker in the past N seconds. A group member may also consider the coordinator broker ‘down’ if no heartbeat response response received in the past N seconds.
```
RebalanceTimeoutSeconds(Int)
```
Time in seconds for each worker to join the group once a rebalance has begun. If the timeout is exceeded, then the worker will be removed from the group, which will cause offset commit failures.
```
HeartbeatRateSeconds(Int)
```
Time in seconds for the member to ‘ping’ the group coordinator. OBS: Care should be taken when picking the number, on one hand, we do not want to flush the broker with requests if we set it too low, on the other hand, if set it too high, it may take too long for the members to realise status changes of the group such as assignment rebalacing or group coordinator switchover etc.
```
MaxRejoinAttempts(Int)
```
Maximum number of times allowed for a member to re-join the group. The gen_server will stop if it reached the maximum number of retries. OBS: ‘let it crash’ may not be the optimal strategy here because the group member id is kept in the gen_server looping state and it is reused when re-joining the group.
```
RejoinDelaySeconds(Int)
```
Delay in seconds before re-joining the group.
```
OffsetCommitIntervalSeconds(Int)
```
The time interval between two OffsetCommitRequest messages.
```
OffsetRetentionSeconds(Int)
```
How long the time is to be kept in kafka before it is deleted. The default special value -1 indicates that the __consumer_offsets topic retention policy is used.

GroupSubscriber

</>

pub type GroupSubscriber {
  GroupSubscriber(name: process.Name(Message))
}

Constructors

GroupSubscriber(name: process.Name(Message))

Message

</>

pub type Message

Values

ack

</>

pub fn ack(cb_state: cb_state) -> CallbackReturn

Acknowledges the processing of the message. Use this in your callback to confirm message receipt.

commit

</>

pub fn commit(cb_state: cb_state) -> CallbackReturn

Commits the offset of the last message that was successfully processed. Use this in your callback to mark messages as processed.

named_client

</>

pub fn named_client(
  name: process.Name(Message),
) -> GroupSubscriber

new

</>

pub fn new(
  name name: process.Name(Message),
  client client: franz.Client,
  group_id group_id: String,
  topics topics: List(String),
  message_type message_type: message_type.MessageType,
  callback callback: fn(franz.KafkaMessage, callback_init_state) -> CallbackReturn,
  init_callback_state init_callback_state: callback_init_state,
) -> Builder(callback_init_state)

Creates a new group subscriber builder. The callback will be called for each message received from the subscribed topics.

start

</>

pub fn start(
  builder: Builder(callback_init_state),
) -> Result(actor.Started(GroupSubscriber), actor.StartError)

Starts a new group subscriber with the configured settings. Returns the process ID of the subscriber on success.

stop

</>

pub fn stop(pid: process.Pid) -> Result(Nil, franz.FranzError)

Stops a running group subscriber.

supervised

</>

pub fn supervised(
  builder: Builder(a),
) -> supervision.ChildSpecification(GroupSubscriber)

This can be used with Gleam’s OTP supervision trees to ensure the subscriber is restarted on failure.

with_consumer_config

</>

pub fn with_consumer_config(
  builder: Builder(callback_init_state),
  consumer_config: config.Config,
) -> Builder(callback_init_state)

Adds a consumer configuration option to the group builder. Multiple configurations can be chained together.

with_group_config

</>

pub fn with_group_config(
  builder: Builder(callback_init_state),
  group_config: Config,
) -> Builder(callback_init_state)

Adds a group configuration option to the group builder. Multiple configurations can be chained together.