Offset Management

Audience: operators and contributors who need to know exactly how EventMesh tracks consumption progress — the two offset layers, per-backend mechanics, the deferred-ACK protocol, and takeover behavior. Part of the offset / load-balancing / frame design trio; see also Load balancing and Frame protocol conversion.

---

Architectural position

EventMesh fully self-manages offsets: no group.id, no reporting to meta. The MQ (RocketMQ / Kafka) is used only as a durable FIFO log; its Consumer Group semantics are never used. The current implementation status of each capability is governed by the capability status table.

The two offset layers

Layer	Meaning	Keyed as	Storage	On takeover
pull offset (how far we pulled from the MQ)	parent#lite@queue (lite) / topic#partition (regular)	storage plugin-local (rocketmq5 = properties file, 5 s atomic checkpoint)	instance dies → new instance replays from head
deliver/ack offset (delivery + ACK progress)	topic#clientId#partition	RocksDBOffsetStore (local)	instance dies → new instance reads -1 → pulls from head → in-flight re-delivered

The two layers are deliberately not merged (different semantics, decoupled): the pull layer lives inside the storage plugin (self-managed cursor), the deliver/ack layer in the runtime's OffsetStore. pullAndDispatchPartition passes startOffset=-1 to storage.poll (the plugin manages its own cursor) — the OffsetStore is not on the takeover path.

The normal path

storage.poll()
  → pull offset self-managed (no commitOffset submitted to the broker)
  → pullAndDispatchPartition (startOffset=-1, plugin-managed cursor)
  → ReliableDispatcher.deliver()
    → Delivery (carries event + mqAckCallback)
    → PushChannel.deliver() → delivered to the client
  → client ACK → ReliableDispatcher.ack()
    → offsetStore.writeOffset() (CAS max — offsets only move forward)
    → storage.ackPulledMessage(topic, popCk)   (RocketMQ 5.x: only now ACK the broker)

Takeover (instance death)

instance dies
  → client reconnects to a new instance via /session/recommend (see load-balancing.md)
  → the new instance re-pulls from the business topic
      · lite:  pollLite replays from the lite head (single cursor, naturally replayable)
      · regular: poll starts from the earliest un-ACKed / head
  → in-flight un-ACKed messages → at-least-once re-delivery (business idempotency covers it)
  → no snapshots written, no offset state migration, no meta reporting

Per-backend differences

Backend	Pull mechanism	ACK mechanism	Restart recovery
RocketMQ 5.x	POP_MESSAGE (broker-managed assignment; partitionCount=-1 → poll-all)	Deferred ACK: poll() does not immediately ACK the broker; ackPulledMessage() fires the broker ACK only after the client ACKs. Crashes lose nothing — the broker's 30 s invisibleTime expires and it re-delivers automatically.	pull offset resumed from the local properties file (5 s checkpoint)
RocketMQ 4.x	PULL_MESSAGE (self-managed cursor, fixed CONSUMER_GROUP, no commitOffset submitted)	No MQ ACK (PULL mode)	pull offset resumed from the local properties file
Kafka	assign + seek (no group.id, ENABLE_AUTO_COMMIT=false)	No MQ ACK	pull offset resumed from the local properties file

Deferred ACK in detail (RocketMQ 5.x, the P2 fix)

The problem: originally poll() immediately called ackNormal(brokerAddr, msg) → the broker marked the message consumed → an EventMesh crash lost the message (at-most-once).

The fixed chain:

① poll()
   → POP_MESSAGE fetches messages
   → does NOT ACK the broker now
   → stores a deferred ACK callback (key = PROPERTY_POP_CK)
   → the EventMeshFrame attribute "empopck" carries the POP check key

② pullAndDispatchPartition()
   → reads frame.attributes().get("empopck")
   → builds mqAck = () -> storage.ackPulledMessage(topic, popCk)
   → dispatcher.deliver(topic, partition, offset, frame, clientId, channel, mqAck)
   → the Delivery stores mqAckCallback

③ client ACK → ReliableDispatcher.ack(deliveryId)
   → offsetStore.writeOffset() (offset advances)
   → delivery.getMqAckCallback().run() → storage.ackPulledMessage(topic, popCk)
   → pendingPopAcks.remove(popCk) → executes the broker ACK_MESSAGE
   → the broker confirms consumption

The at-least-once guarantee: if EventMesh crashes between ① and ③, the broker's invisibleTime (30 s) expires → automatic re-delivery → a new instance pulls and processes again.

Monotonic offset writes (the P4 fix)

The problem: writeOffset used a plain overwrite (set(offset)) — after a restart, a fast consumer group's offset could be dragged down by a slow group's replayed messages.

The fix:

• InMemoryOffsetStore: AtomicLong.accumulateAndGet(offset, Math::max) — CAS-max.
• RocksDBOffsetStore: read current → skip if offset <= current → else put — a conditional write.

Retired designs

• MetaBackedOffsetStore (reporting a million keys to meta every second) is off by default — opt-in via -Deventmesh.offset.meta=true; the class is kept in reserve.
• No group.id, ever — the EventMesh-owned-offset rule stands.

Open item

• RocketMQ 4.x ConsumeQueue cleanup — commitOffset remains a no-op. Verified in practice as a false alarm for storage growth (the CommitLog is time-GC'd via fileReservedTime and ConsumeQueues are trimmed in lockstep), but worth re-checking per deployment.

Where the code lives

Piece	Location
Offset stores	eventmesh-runtime/.../offset/RocksDBOffsetStore.java, InMemoryOffsetStore.java
Deferred POP ACK	eventmesh-storage-plugin/eventmesh-storage-rocketmq5/ (pendingPopAcks, ackPulledMessage)
ACK-driven advance	eventmesh-runtime/.../delivery/ReliableDispatcher.java (ack())
Boot alignment	UniRuntime.alignPullOffsetsToAck
Tests	OffsetMonotonicAndRecoveryTest, DeferredAckDispatcherTest, AckTimeoutRedeliveryIntegrationTest