Page History

...

• on discovering newly created topics or newly added partitions to existing topics, the controller sends an error code in the HeartbeatResponse to all alive group members and waits for a new RegisterConsumer request from each consumer, except the one that sent the initial RegisterConsumer.
• on receiving an error code in the HeartbeatResponse, all consumers stop fetching, commit offsets, re-discover the controller through list_groups(my_group) and send a RegisterConsumer request to the controller
• on receiving RegisterConsumer from all consumers of the group membership change, the controller sends the RegisterConsumerResponse to all the new group members, with the new partitions and the respective offsets to restart consumption from. For newly added partitions and topics, the offset is set to smallest (0L).
• on receiving the RegisterConsumerResponse the consumer is now able to start consuming its partitions and must now start sending heartbeat messages back to the controller with the current generation id.

Offset Rewind

 while(true) {  List<MessageAndMetadata> messages = consumer.poll(timeout);
  process(messages);
  if(rewind_required) {
     List<PartitionOffset> partitionOffsets = new …;
     partitionOffsets.add(new PartitionOffset(topic, partition, offset);
     rewind_offsets(partitionOffsets);     
  }  
}

the consumer stops fetching data, sends a RewindConsumer request to the controller and awaits a RewindConsumerResponseon receiving a RewindConsumer request, the controller sends an error code in the HeartbeatResponse to the current owners of the rewound partitionson receiving an error code in the HeartbeatResponse, the affected consumers stop fetching, commit offsets and send the RegisterConsumer request to the controlleron receiving a RegisterConsumer request from the affected members of the group, the controller records the new offsets for the specified partitions and sends the new offsets to the respective consumers.on receiving a RegisterConsumerResponse, the consumers start fetching data from the specified offsets

Low-level RPC API

TBD we need to design an underlying client network abstraction that can be used to send requests. This should ideally handle metadata and addressing by broker id, multiplex over multiple connections, and support both blocking and non-blocking operations.

...