...

and enabling staleness-based detection of stuck workers. 

1.1 Why this matters now

These problems are blocking adoption of share groups for the workloads where they would matter most:

...

difference between "broker knows work is happening" and "broker silently holds the lock" determines whether the system can be operated with confidence.

2 Public Interfaces

2.1 New client API

public interface ShareConsumer<K, V> {
    void markInProgress(ConsumerRecord<K, V> record);
}

Why: explicit, payload-carrying signal that real work is happening (vs RENEW's payload-less timer reset).

2.2 New RecordState value

RecordState.IN_PROGRESS  ((byte) 5)

...

Why: preserves the invariant "client must hold the lock before claiming work." All terminal/release paths from ACQUIRED are mirrored.

2.3 New configs

2.4 New wire RPC

MarkInProgressRequest/Response (apiKey 81). Schema mirrors ShareAcknowledgeRequest with ProgressBatches carrying (firstOffset, lastOffset).

Why a separate RPC, not a new AcknowledgeType: progress is conceptually distinct from acknowledgement.

2.5 New admin API

admin.describeShareGroupInFlight(groupId, options)
   .heldLongerThanMs(...)
   .progressStalenessMs(...)
   .onlyState(IN_PROGRESS)

...

Why: same operator visibility on the CLI surface, including a STALE annotation that flags problem records.

3 Proposed Changes

3.1 InFlightState extension

Add four fields to SharePartition.InFlightState:

lastProgressTimeMs       (-1 if never markInProgress'd)
inProgressStartTimeMs    (when transitioned to IN_PROGRESS)

3.2 markInProgress handler

In SharePartitionManager.handleMarkInProgressRequest:

 1.  Validate ownership (memberId matches) and current state (ACQUIRED or IN_PROGRESS).
 2.  Validate cumulative IN_PROGRESS time ≤ max.lock.extension.ms.
 3.  Transition ACQUIRED → IN_PROGRESS (or refresh lastProgressTimeMs if already in IN_PROGRESS).
 4.  Cancel + reschedule AcquisitionLockTimerTask.
 5.  Async-write delta record to __share_group_state.

Why: minimal addition; reuses the existing writeLock + timer + persistence machinery from acknowledge().

3.3 Staleness sweeper

A background task per SharePartition runs every (staleness.threshold.ms / 4):

•  If lastProgressTimeMs is older than threshold → release to AVAILABLE, increment deliveryCount.
•  If inProgressStartTimeMs exceeds max.extension.ms → release regardless.

Why: broker is the only authoritative detector of "no signal in N seconds" — clients that need detection are the ones that have crashed.

Rebalance flow

In SharePartitionManager.handleMemberLeaving, distinguish:

•  ACQUIRED records → release immediately (existing behaviour).
•  IN_PROGRESS records with fresh progress → hold for group.share.in.progress.rebalance.grace.ms (default 5000) before releasing.
•  IN_PROGRESS records with stale progress → release immediately (worker was already stuck).

Why: preserves work for healthy in-progress records across brief member churn.

DelayedShareFetch

acquirablePartitions already excludes ACQUIRED; extend to also exclude IN_PROGRESS. 

Client side

ShareConsumerImpl.markInProgress enqueues a pending progress batch; flushed in a separate MarkInProgress RPC on the next poll() boundary

or when the pending count exceeds max.in-flight.progress.batches (default 100).
Why: amortises RPC cost; mirrors how acknowledge() is piggybacked today.