You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "wellkilo (via GitHub)" <gi...@apache.org> on 2026/09/29 07:31:40 UTC
[PR] [runtime][java] Surface Kafka action-state send failures [flink-agents]
wellkilo opened a new pull request, #1177:
URL: https://github.com/apache/flink-agents/pull/1177
<!-- Please add the relevant components in the PR title. -->
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
attempt-local cache changes, 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 waits on the `Future<RecordMetadata>` returned by that specific `send()` call.
3. After the acknowledgement succeeds, it flushes the producer and updates the attempt-local
cache.
4. Only then does `put(...)` return successfully.
#### Key decisions
- Wait on the send future instead of relying on `flush()`. 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.
- 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 / acknowledgement / flush outcome | Result |
| --- | --- |
| `send()` throws synchronously | `RuntimeException`; cache unchanged |
| Send future completes exceptionally | `RuntimeException` with the broker failure as root cause; flush and cache update are skipped |
| Acknowledgement succeeds, `flush()` fails | `RuntimeException`; cache update is skipped |
| Acknowledgement wait is interrupted | `InterruptedException`; interrupt flag restored; flush and cache update are skipped |
| Acknowledgement and flush succeed | Cache is updated, 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.
- An asynchronous acknowledgement failure is propagated to the caller and does not populate the
attempt-local cache.
- An interrupted acknowledgement wait preserves `Thread.currentThread().isInterrupted()` and
throws `InterruptedException`.
- The method does not update the cache on send, acknowledgement, or flush failure.
#### 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 leaves the cache unchanged | `testPutPropagatesAsyncBrokerFailureBeforeCachingState` |
| Interruption is propagated, the interrupt flag is restored, and the cache is unchanged | `testPutPreservesInterruptWhileWaitingForBrokerAcknowledgement` |
| The successful send, flush, 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.
<details>
<summary>Implementation invariants and supporting evidence</summary>
- `actionStates.put(...)` is reached only after `future.get()` and `producer.flush()` 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.
</details>
### 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`
- [x] `doc-included`
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
Generated-by: TraeCode (GPT-5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Surface Kafka action-state send failures [flink-agents]
Posted by "sangkyoonnam (via GitHub)" <gi...@apache.org>.
sangkyoonnam commented on code in PR #1177:
URL: https://github.com/apache/flink-agents/pull/1177#discussion_r4254630230
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -212,6 +214,22 @@ public KafkaActionStateStore(
LOG.info("Initialized KafkaActionStateStore with topic: {}", topic);
}
+ /**
+ * Persist the given action state to Kafka and add it to the attempt-local cache.
+ *
+ * <p>The record is sent, the producer is flushed, and then the record's send future is awaited.
+ * Flushing before the wait forces the record out of the accumulator immediately, so a
+ * synchronous write is not delayed by the producer's {@code linger.ms}. The send future is
+ * still awaited afterwards because a successful {@code flush()} does not guarantee that an
+ * individual record was acknowledged: a record-level, asynchronous broker failure is surfaced
+ * to the caller.
+ *
+ * <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.
Review Comment:
"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:
```suggestion
* <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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Surface Kafka action-state send failures [flink-agents]
Posted by "wenjin272 (via GitHub)" <gi...@apache.org>.
wenjin272 commented on PR #1177:
URL: https://github.com/apache/flink-agents/pull/1177#issuecomment-6116842966
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:
```java
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()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Surface Kafka action-state send failures [flink-agents]
Posted by "sangkyoonnam (via GitHub)" <gi...@apache.org>.
sangkyoonnam commented on code in PR #1177:
URL: https://github.com/apache/flink-agents/pull/1177#discussion_r4235328733
##########
runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java:
##########
@@ -224,13 +224,16 @@ public void put(Object key, long seqNum, Action action, Event event, ActionState
try {
ProducerRecord<String, ActionState> kafkaRecord =
new ProducerRecord<>(topic, stateKey, state);
- producer.send(kafkaRecord);
- actionStates.put(stateKey, state);
+ producer.send(kafkaRecord).get();
producer.flush();
Review Comment:
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:
```java
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.
##########
runtime/API.md:
##########
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [runtime][java] Surface Kafka action-state send failures [flink-agents]
Posted by "wellkilo (via GitHub)" <gi...@apache.org>.
wellkilo commented on PR #1177:
URL: https://github.com/apache/flink-agents/pull/1177#issuecomment-6117032659
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org