[wellkilo]

Linked issue: #1173

Purpose of change

A caller no longer receives a successful return from KafkaActionStateStore.put(...) when Kafka
later rejects the state record. The asynchronous broker failure now fails the task before the
update is added to the attempt-local cache, preventing a checkpoint from treating an action-state
update as durable when its Kafka record was never acknowledged.

Runtime flow:

1. put(...) encodes the composite state key and sends the ProducerRecord.
2. It flushes the producer, which forces the record out of the accumulator immediately so the
   synchronous write is not delayed by the producer's linger.ms.
3. It waits on the Future<RecordMetadata> returned by that specific send() call. The wait is
   still needed because flush() can return normally while the per-record future retains an
   asynchronous failure.
4. Only after the flush and the acknowledgement succeed does it update the attempt-local cache and
   return successfully.

Key decisions

• Flush before awaiting the send future, so a synchronous write is not delayed by the producer's
  linger.ms (5 ms by default since kafka-clients 4.0). The wait is still performed afterwards
  because Kafka's MockProducer demonstrates that flush() can return normally while the
  per-record future retains an asynchronous failure.
• Keep the existing synchronous persistence contract: successful return still means both the
  acknowledgement and flush completed.
• Re-throw InterruptedException with the thread's interrupt flag restored, rather than wrapping
  it as a generic Kafka persistence failure. An interrupt raised while blocked in flush() still
  surfaces as Kafka's InterruptException wrapped in the generic RuntimeException with the flag
  restored, matching the previous behavior.
• Do not add a new acknowledgement timeout or configuration. This patch only makes the existing
  wait observable; it does not change the configured producer behavior.

Interaction decisions

Send / flush / acknowledgement outcome	Result
send() throws synchronously	RuntimeException; the state is not added to the cache
flush() fails	RuntimeException; the state is not added to the cache
Send future completes exceptionally	RuntimeException with the broker failure as root cause; the state is not added to the cache
Acknowledgement wait is interrupted	InterruptedException; interrupt flag restored; the state is not added to the cache
Flush and acknowledgement succeed	The state is added to the cache, then put(...) returns normally

Behavioral contracts

• A successful put(...) means the specific record was acknowledged, the producer was flushed,
  and the attempt-local cache contains the state for this update.
• An asynchronous acknowledgement failure is propagated to the caller; the update is not added to
  the attempt-local cache.
• An interrupted acknowledgement wait preserves Thread.currentThread().isInterrupted() and
  throws InterruptedException.
• The method does not add the update to the cache on send, flush, or acknowledgement failure. It
  makes no claim about the rest of the cache, which callers can mutate through previously cached
  ActionState references.

Failure behavior

Kafka send and flush exceptions are surfaced as
RuntimeException("Failed to send action state to Kafka", cause). Interruption is not wrapped:
it is re-thrown with the interrupt status preserved. A null producer retains the pre-existing
behavior of logging and returning.

Tests

Behavioral contract	Evidence
Asynchronous broker failure fails put(...) and does not cache the state	testPutPropagatesAsyncBrokerFailureBeforeCachingState
Interruption is propagated, the interrupt flag is restored, and the state is not cached	testPutPreservesInterruptWhileWaitingForBrokerAcknowledgement
The successful send, flush, acknowledgement, and cache path remains intact	testPutActionState and the existing KafkaActionStateStoreTest suite

Local verification completed on the branch: KafkaActionStateStoreTest (51/51), repository
lint/Spotless, and Apache RAT all pass.

Not verified: live-broker failover/recovery and a never-completing broker acknowledgement. The
regression uses MockProducer to make the record-level failure deterministic; no acknowledgement
timeout is introduced by this change.

Implementation invariants and supporting evidence

• actionStates.put(...) is reached only after producer.flush() and future.get() complete
  successfully.
• The InterruptedException catch restores the flag before re-throwing the same exception.
• The existing ActionStateStore.put(...) signature already declares throws Exception, so no
  public interface change is required.
• The Kafka record schema and the attempt-local cache key remain unchanged.

API

No public API signature, event schema, checkpoint format, or configuration change is introduced.
The caller-visible behavior change is limited to propagating failures that were previously hidden
by ignoring the send future.

Documentation

• doc-needed
• doc-not-needed
• doc-included

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: TraeCode (GPT-5)
Generated-by: Codex CLI 0.159.0 (GPT-5)

[sangkyoonnam]

I'd flush before waiting on the send future. As written, a put with a small record in an otherwise idle batch waits out the 5 ms default linger.ms before the record is sent. Awaiting the record's future is the right fix otherwise: it matches FlussActionStateStore.put, which calls writer.append(row).get() before caching, and both new tests fail when I revert only KafkaActionStateStore.java (2 failures in KafkaActionStateStoreTest, 51/51 with the change). Separately, a non-blocking suggestion on the two new Markdown files. Happy to rerun the timing on the revised branch.

[sangkyoonnam]

For small records in otherwise idle batches, waiting on get() before flush() adds roughly the default linger delay. Puts are sequential, so the next put can't fill the batch while this one waits. kafka-clients 4.0 changed the linger.ms default from 0 to 5 ms, and createProducerProp() doesn't set it. The existing flush() is what bypassed linger. After get() returns, it no longer helps this record. FlussActionStateStore sets its writer batch timeout to zero for the same reason (L156-L158).

I timed the three orderings with a plain KafkaProducer (kafka-clients 4.0.0, acks=all, retries=3, string payloads) against a local apache/kafka-native:3.8.0 container. 200 sequential sends per ordering, second of two rounds, ms per send iteration:

main  send; flush         0.50
PR    send.get; flush     6.88
alt   send; flush; get    0.84

A fresh action writes initial and completion state, and durable calls can add more synchronous writes, so the delay repeats per action. Flushing first bypasses linger, and get() still checks the record's result before the cache insert:

Future<RecordMetadata> ack = producer.send(kafkaRecord);
producer.flush();
ack.get();
actionStates.put(stateKey, state);

Setting linger.ms to 0 in createProducerProp() would also work and keeps your order. With the snippet, the broker-failure test still passes. The interrupt test fails only on assertThat(flushCalled).isFalse() at L277; flipping it to isTrue() keeps the ordering pinned. An interrupt while blocked in flush() produces Kafka's InterruptException wrapped in the generic RuntimeException, with the flag restored. That matches main today. Surfacing InterruptedException from that path would need a catch that translates Kafka's exception. I haven't tested a real interrupt. The flow and interaction table in the PR description would need the same update.

[sangkyoonnam]

Non-blocking. These two files repeat what the PR description already says, and nothing in the repo links to them. I'd remove them. If the contract needs a permanent home, the Javadoc on KafkaActionStateStore.put keeps it next to the code: for an initialized store, a normal return means the record was acknowledged and cached.

[wenjin272]

Thanks for the fix! The new tests correctly catch the previously ignored send failure.

I agree with @sangkyoonnam that we should flush before waiting on the send future, so each synchronous state write avoids Kafka 4.0’s default linger delay:

Future<RecordMetadata> ack = producer.send(kafkaRecord);
producer.flush();
ack.get();
actionStates.put(stateKey, state);

Could we also move the relevant contract into put()’s Javadoc and remove the two Markdown files? The wording should avoid promising that the cache remains unchanged on failure, since callers can already mutate a cached ActionState before calling put().

[wellkilo]

Thanks @sangkyoonnam and @wenjin272 — agreed on all three points, and the timing data is convincing. Addressed in b66fd26.

Ordering. Reordered to send(); flush(); ack.get(); so flush() still bypasses the linger delay (createProducerProp() doesn't set linger.ms, and 4.0 defaults it to 5 ms). ack.get() still runs before the cache insert, so a record-level async failure is still surfaced. I kept the reorder instead of setting linger.ms = 0, to avoid a producer-config change and stay close to main.

Interrupts. Flipped the interrupt test's assertThat(flushCalled) to isTrue() to pin the ordering. An interrupt while blocked in flush() keeps main's behavior (Kafka's InterruptException wrapped in the generic RuntimeException, flag restored); the new get() path rethrows InterruptedException with the flag restored. I haven't tested a real interrupt either.

Docs. Removed runtime/API.md and runtime/Method.md, and moved the contract onto KafkaActionStateStore.put(...)'s Javadoc. Dropped the "cache unchanged" wording — the Javadoc now only says the state isn't added to the cache when send/ack/flush fails, which is what put() actually controls.

The PR description's flow and interaction table are updated to match the new order.

[sangkyoonnam]

The send, flush, then ack.get() order and the interrupt handling look right to me, and KafkaActionStateStoreTest passes at 7385011 (51 tests). One wording nit on the put() Javadoc, inline.

[sangkyoonnam]

"The state is not cached" can still read as a claim about what the cache holds, which is the concern wenjin272 raised. DurableExecutionManager.maybePersistTaskResult takes the cached ActionState, marks it completed, and passes that same object to put(). If that put() fails, the cache still holds the state, already marked completed. Describing only what put() does avoids that:

Suggested change

	* <p>The state is added to the attempt-local cache only after both the flush and the
	* acknowledgement succeed. If the send, the acknowledgement, or the flush fails, the state is
	* not cached and a {@link RuntimeException} wrapping the cause is thrown. If the
	* acknowledgement wait is interrupted, {@link InterruptedException} is rethrown with the
	* thread's interrupt flag restored.
	* <p>{@code put()} adds the state to the attempt-local cache only after both the flush and the
	* acknowledgement succeed. If the send, the flush, or the acknowledgement fails, {@code put()}
	* does not add the state and throws a {@link RuntimeException} wrapping the cause. If the
	* acknowledgement wait is interrupted, {@link InterruptedException} is rethrown with the
	* thread's interrupt flag restored.

I applied it locally and tools/lint.sh -c passes on JDK 11.