Bug: ConsumerGroup ignores current offset

In rare cases the consumer (after the intial startup) doesn't pick up the latest offset from the consumer group for the given partition. Furthermore, if the topic has a cleanup policy, like deleting messages older then 7 days, the consumer group has a lag and isn't setting the latest non deleted offset. The group is just stuck at offset 0 and isn't receiving new incoming messages! Screenshot from panda console with a lag (retention policy delete, all messages deleted but offset is 0 instead of 22):  Log after startup: ``` 2025-04-23T21:35:15.755519Z INFO samsa::consumer_group: Member b"samsa-c578b962-bdca-40e4-9825-e3904feb0934" | Heartbeat 2025-04-23T21:35:15.756615Z INFO commit_offset{correlation_id=1 client_id="samsa" group_id="ism" coordinator_conn=TcpConnection { stream: PollEvented { io: Some(TcpStream { addr: [::1]:63146, peer: [::1]:19192, socket: 804 }) } } generation_id=8 member_id=b"samsa-c578b962-bdca-40e4-9825-e3904feb0934" offsets={("user-notification-events", 0): 0} retention_time_ms=100000}: samsa::consumer: Member b"samsa-c578b962-bdca-40e4-9825-e3904feb0934" - Committing offsets {("user-notification-events", 0): 0} 2025-04-23T21:35:15.763161Z DEBUG consume{self=Consumer { cluster_metadata: ClusterMetadata { connection_params: [BrokerAddress { host: "localhost", port: 19192 }], broker_connections: {1: TcpConnection { stream: PollEvented { io: Some(TcpStream { addr: [::1]:63159, peer: [::1]:19192, socket: 832 }) } }}, brokers: [Broker { node_id: 1, host: b"localhos t", port: 19192, rack: None }], topics: [Topic { error_code: None, name: b"user-notification-events", is_internal: false, partitions: [Partition { error_code: None, partition_index: 0, leader_id: 1, replica_nodes: [1], isr_nodes: [1] }] }], correlation_id: 1, client_id: "samsa", topic_names: ["user-notification-events"], controller_id: 1 }, fetch_params: FetchParams { correlation_id: 1, client_id: "samsa", max_wait_ms: 200, min_bytes: 100, max_bytes: 30000, max_partition_bytes: 20000, isolation_level: 0 }, assigned_topic_partitions: {"user-notification-events": [0]}, offsets: {("user-notification-events", 0): 0} }}: samsa::metadata: Broker 1 is in charge of {"user-notification-events": [0]} 2025-04-23T21:35:15.763353Z DEBUG consume{self=Consumer { cluster_metadata: ClusterMetadata { connection_params: [BrokerAddress { host: "localhost", port: 19192 }], broker_connections: {1: TcpConnection { stream: PollEvented { io: Some(TcpStream { addr: [::1]:63159, peer: [::1]:19192, socket: 832 }) } }}, brokers: [Broker { node_id: 1, host: b"localhos t", port: 19192, rack: None }], topics: [Topic { error_code: None, name: b"user-notification-events", is_internal: false, partitions: [Partition { error_code: None, partition_index: 0, leader_id: 1, replica_nodes: [1], isr_nodes: [1] }] }], correlation_id: 1, client_id: "samsa", topic_names: ["user-notification-events"], controller_id: 1 }, fetch_param s: FetchParams { correlation_id: 1, client_id: "samsa", max_wait_ms: 200, min_bytes: 100, max_bytes: 30000, max_partition_bytes: 20000, isolation_level: 0 }, assigned_topic_partitions: {"user-notification-events": [0]}, offsets: {("user-notification-events", 0): 0} }}:fetch{broker_conn=TcpConnection { stream: PollEvented { io: Some(TcpStream { addr: [: :1]:63159, peer: [::1]:19192, socket: 832 }) } } correlation_id=1 client_id="samsa" max_wait_ms=200 min_bytes=100 max_bytes=30000 max_partition_bytes=20000 isolation_level=0 topic_partitions={"user-notification-events": [0]} offsets={("user-notification-events", 0): 0}}: samsa::consumer: Consuming {"user-notification-events": [0]} with offsets {("user-notification-events", 0): 0} ``` After I have sent new events to the topic, the lag grows and the consumer is ignoring newer events. It is stuck at the 0 offset.  If you want to preproduce it: 1. Send some events, delete them with retention policy or manual, start the consumer group -> nothing is happening 2. Send again some events, they will not be processed. The other case where the offset is sometimes resetted to zero is hard to reproduce, its more often with Redpanda than Kafka. But I get it regularly when I start my consumer 10-20x in a row.