elsa v0.12.3 Elsa.Group.Acknowledger

Handles acknowledgement of messages to the group coordinator to prevent the group manager from queuing up messages for acknowledgement when events such as a rebalance occur.

Link to this section Summary

Functions

ack(connection, map)

Trigger acknowledgement of processed messages back to the cluster.

ack(connection, topic, partition, generation_id, offset)

Trigger acknowledgement of processed messages back to the cluster.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

get_latest_offset(acknowledger, topic, partition)

Retrieve the latest offset for a topic and partition. Primarily used for reinitializing consumer workers to the latest unacknowledged offset after a rebalance or other disruption.

start_link(opts)

Instantiate an acknowledger process and register it to the Elsa registry.

update_generation_id(acknowledger, generation_id)

Sync the group generation ID back to the acknowledger state for validation.

Link to this section Functions

ack(connection, map)

ack(Elsa.connection(), %{
  topic: Elsa.topic(),
  partition: Elsa.partition(),
  generation_id: Elsa.Group.Manager.generation_id(),
  offset: integer()
}) :: :ok

Trigger acknowledgement of processed messages back to the cluster.

ack(connection, topic, partition, generation_id, offset)

ack(
  Elsa.connection(),
  Elsa.topic(),
  Elsa.partition(),
  Elsa.Group.Manager.generation_id(),
  integer()
) :: :ok

Trigger acknowledgement of processed messages back to the cluster.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

get_latest_offset(acknowledger, topic, partition)

get_latest_offset(GenServer.server(), Elsa.topic(), Elsa.partition()) ::
  Elsa.Group.Manager.begin_offset() | nil

Retrieve the latest offset for a topic and partition. Primarily used for reinitializing consumer workers to the latest unacknowledged offset after a rebalance or other disruption.

start_link(opts)

start_link(term()) :: GenServer.on_start()

Instantiate an acknowledger process and register it to the Elsa registry.

update_generation_id(acknowledger, generation_id)

update_generation_id(GenServer.server(), Elsa.Group.Manager.generation_id()) ::
  :ok

Sync the group generation ID back to the acknowledger state for validation.