This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/main by this push:
new d9bb5075 [hotfix][runtime] Aggregate close failures across the
Exception/Error boundary (#974)
d9bb5075 is described below
commit d9bb5075c909fb188054cc8ee5e2674efbf31b5a
Author: Weiqing Yang <[email protected]>
AuthorDate: Sat Aug 8 22:17:20 2026 -0700
[hotfix][runtime] Aggregate close failures across the Exception/Error
boundary (#974)
Generated-by: Claude Code (claude-opus-5)
---
.../runtime/actionstate/FlussActionStateStore.java | 22 +++-
.../runtime/actionstate/KafkaActionStateStore.java | 20 ++--
.../actionstate/FlussActionStateStoreTest.java | 115 +++++++++++++++++++++
.../actionstate/KafkaActionStateStoreTest.java | 57 ++++++++++
4 files changed, 200 insertions(+), 14 deletions(-)
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java
index b6993281..0a20fe2b 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStore.java
@@ -21,6 +21,7 @@ import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.plan.AgentConfiguration;
import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.fluss.client.Connection;
import org.apache.fluss.client.ConnectionFactory;
@@ -490,15 +491,28 @@ public class FlussActionStateStore implements
ActionStateStore {
@Override
public void close() throws Exception {
- try {
- if (table != null) {
+ // Catching Throwable rather than Exception is what keeps the
connection close reachable
+ // when the table close fails with an Error, and what keeps that first
failure the one the
+ // caller sees — a later connection failure rides along as suppressed
instead of replacing
+ // it.
+ Throwable firstException = null;
+ if (table != null) {
+ try {
table.close();
+ } catch (Throwable t) {
+ firstException = t;
}
- } finally {
- if (connection != null) {
+ }
+ if (connection != null) {
+ try {
connection.close();
+ } catch (Throwable t) {
+ firstException = ExceptionUtils.firstOrSuppressed(t,
firstException);
}
}
+ if (firstException != null) {
+ ExceptionUtils.rethrowException(firstException);
+ }
}
private void maybeCreateDatabaseAndTable() {
diff --git
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java
index ea7ea146..99519acb 100644
---
a/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java
+++
b/runtime/src/main/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStore.java
@@ -22,6 +22,7 @@ import org.apache.flink.agents.api.Event;
import org.apache.flink.agents.plan.AgentConfiguration;
import org.apache.flink.agents.plan.actions.Action;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
@@ -331,27 +332,26 @@ public class KafkaActionStateStore implements
ActionStateStore {
@Override
public void close() throws Exception {
- Exception firstException = null;
+ // Catching Throwable rather than Exception is what keeps the consumer
close reachable when
+ // the producer close fails with an Error, and what keeps that first
failure the one the
+ // caller sees — a later consumer failure rides along as suppressed
instead of replacing it.
+ Throwable firstException = null;
if (producer != null) {
try {
producer.close();
- } catch (Exception e) {
- firstException = e;
+ } catch (Throwable t) {
+ firstException = t;
}
}
if (consumer != null) {
try {
consumer.close();
- } catch (Exception e) {
- if (firstException == null) {
- firstException = e;
- } else {
- firstException.addSuppressed(e);
- }
+ } catch (Throwable t) {
+ firstException = ExceptionUtils.firstOrSuppressed(t,
firstException);
}
}
if (firstException != null) {
- throw firstException;
+ ExceptionUtils.rethrowException(firstException);
}
}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java
index 308ad3d4..f6ba5fcc 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/FlussActionStateStoreTest.java
@@ -36,7 +36,9 @@ import java.util.concurrent.CompletableFuture;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.catchThrowable;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -157,6 +159,7 @@ public class FlussActionStateStoreTest {
assertThat(actionStates).isEmpty();
}
+ /** Contract: both the table and the connection are closed when neither
close fails. */
@Test
void testCloseClosesResources() throws Exception {
Table mockTable = mock(Table.class);
@@ -170,4 +173,116 @@ public class FlussActionStateStoreTest {
verify(mockTable).close();
verify(mockConnection).close();
}
+
+ /**
+ * Contract: a recorded table-close failure survives a clean connection
close — the table's
+ * exception still reaches the caller, and the connection is closed.
+ */
+ @Test
+ void testCloseClosesConnectionWhenTableCloseFails() throws Exception {
+ Table failingTable = mock(Table.class);
+ Connection mockConnection = mock(Connection.class);
+ RuntimeException tableFailure = new RuntimeException("table close
failed");
+ doThrow(tableFailure).when(failingTable).close();
+
+ FlussActionStateStore closeableStore =
+ new FlussActionStateStore(actionStates, mockConnection,
failingTable, mockWriter);
+
+
assertThat(catchThrowable(closeableStore::close)).isSameAs(tableFailure);
+
+ verify(mockConnection).close();
+ }
+
+ /**
+ * Contract: when both closes fail, the table's exception is the one
thrown and the connection's
+ * is attached to it as a suppressed exception, so neither failure is lost.
+ */
+ @Test
+ void testCloseKeepsTableFailureWhenBothCloseFail() throws Exception {
+ Table failingTable = mock(Table.class);
+ Connection failingConnection = mock(Connection.class);
+ IOException tableFailure = new IOException("table close failed");
+ IOException connectionFailure = new IOException("connection close
failed");
+ doThrow(tableFailure).when(failingTable).close();
+ doThrow(connectionFailure).when(failingConnection).close();
+
+ FlussActionStateStore closeableStore =
+ new FlussActionStateStore(
+ actionStates, failingConnection, failingTable,
mockWriter);
+
+ Throwable thrown = catchThrowable(closeableStore::close);
+
+ assertThat(thrown).isSameAs(tableFailure);
+ assertThat(thrown.getSuppressed()).containsExactly(connectionFailure);
+ }
+
+ /**
+ * Contract: when only the connection close fails, its exception reaches
the caller unchanged,
+ * with nothing attached as suppressed.
+ */
+ @Test
+ void testCloseThrowsConnectionFailureWhenOnlyConnectionCloseFails() throws
Exception {
+ Table mockTable = mock(Table.class);
+ Connection failingConnection = mock(Connection.class);
+ RuntimeException connectionFailure = new RuntimeException("connection
close failed");
+ doThrow(connectionFailure).when(failingConnection).close();
+
+ FlussActionStateStore closeableStore =
+ new FlussActionStateStore(actionStates, failingConnection,
mockTable, mockWriter);
+
+ Throwable thrown = catchThrowable(closeableStore::close);
+
+ assertThat(thrown).isSameAs(connectionFailure);
+ assertThat(thrown.getSuppressed()).isEmpty();
+ }
+
+ /**
+ * Contract: when closing the table throws a non-{@code Exception} {@code
Throwable} and the
+ * connection close then fails too, the throwable stays the failure the
caller sees, the
+ * connection is still closed, and the connection's failure is attached as
suppressed.
+ */
+ @Test
+ void testCloseKeepsTableErrorWhenConnectionCloseAlsoFails() throws
Exception {
+ Table failingTable = mock(Table.class);
+ Connection failingConnection = mock(Connection.class);
+ NoClassDefFoundError tableFailure = new
NoClassDefFoundError("simulated teardown failure");
+ RuntimeException connectionFailure = new RuntimeException("connection
close failed");
+ doThrow(tableFailure).when(failingTable).close();
+ doThrow(connectionFailure).when(failingConnection).close();
+
+ FlussActionStateStore closeableStore =
+ new FlussActionStateStore(
+ actionStates, failingConnection, failingTable,
mockWriter);
+
+ Throwable thrown = catchThrowable(closeableStore::close);
+
+ assertThat(thrown).isSameAs(tableFailure);
+ assertThat(thrown.getSuppressed()).containsExactly(connectionFailure);
+ verify(failingConnection).close();
+ }
+
+ /**
+ * Contract: when the table close fails and the connection close then
throws a non-{@code
+ * Exception} {@code Throwable}, the table's exception stays the failure
the caller sees and the
+ * connection's throwable is attached as suppressed.
+ */
+ @Test
+ void testCloseKeepsTableFailureWhenConnectionCloseThrowsError() throws
Exception {
+ Table failingTable = mock(Table.class);
+ Connection failingConnection = mock(Connection.class);
+ IOException tableFailure = new IOException("table close failed");
+ NoClassDefFoundError connectionFailure =
+ new NoClassDefFoundError("simulated teardown failure");
+ doThrow(tableFailure).when(failingTable).close();
+ doThrow(connectionFailure).when(failingConnection).close();
+
+ FlussActionStateStore closeableStore =
+ new FlussActionStateStore(
+ actionStates, failingConnection, failingTable,
mockWriter);
+
+ Throwable thrown = catchThrowable(closeableStore::close);
+
+ assertThat(thrown).isSameAs(tableFailure);
+ assertThat(thrown.getSuppressed()).containsExactly(connectionFailure);
+ }
}
diff --git
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java
index c0a01bd0..1d8ae231 100644
---
a/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java
+++
b/runtime/src/test/java/org/apache/flink/agents/runtime/actionstate/KafkaActionStateStoreTest.java
@@ -39,6 +39,7 @@ import java.util.Map;
import static
org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy.EARLIEST;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -335,6 +336,62 @@ public class KafkaActionStateStoreTest {
assertThat(thrown.getSuppressed()).isEmpty();
}
+ /**
+ * Contract: when closing the producer throws a non-{@code Exception}
{@code Throwable}, the
+ * consumer is still closed and the throwable reaches the caller unchanged.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ void testCloseClosesConsumerWhenProducerCloseThrowsError() {
+ Producer<String, ActionState> failingProducer = mock(Producer.class);
+ Consumer<String, ActionState> consumer = mock(Consumer.class);
+ NoClassDefFoundError producerFailure =
+ new NoClassDefFoundError("simulated teardown failure");
+ doThrow(producerFailure).when(failingProducer).close();
+
+ KafkaActionStateStore store =
+ new KafkaActionStateStore(
+ actionStates,
+ new AgentConfiguration(),
+ failingProducer,
+ consumer,
+ TEST_TOPIC);
+
+ assertThat(catchThrowable(store::close)).isSameAs(producerFailure);
+
+ verify(consumer).close();
+ }
+
+ /**
+ * Contract: when the producer close fails and the consumer close then
throws a non-{@code
+ * Exception} {@code Throwable}, the producer's exception stays the
failure the caller sees and
+ * the consumer's throwable is attached as suppressed.
+ */
+ @Test
+ @SuppressWarnings("unchecked")
+ void testCloseKeepsProducerFailureWhenConsumerCloseThrowsError() {
+ Producer<String, ActionState> failingProducer = mock(Producer.class);
+ Consumer<String, ActionState> failingConsumer = mock(Consumer.class);
+ RuntimeException producerFailure = new RuntimeException("producer
close failed");
+ NoClassDefFoundError consumerFailure =
+ new NoClassDefFoundError("simulated teardown failure");
+ doThrow(producerFailure).when(failingProducer).close();
+ doThrow(consumerFailure).when(failingConsumer).close();
+
+ KafkaActionStateStore store =
+ new KafkaActionStateStore(
+ actionStates,
+ new AgentConfiguration(),
+ failingProducer,
+ failingConsumer,
+ TEST_TOPIC);
+
+ Throwable thrown = catchThrowable(store::close);
+
+ assertThat(thrown).isSameAs(producerFailure);
+ assertThat(thrown.getSuppressed()).containsExactly(consumerFailure);
+ }
+
/** Contract: both the producer and the consumer are closed when neither
close fails. */
@Test
void testCloseClosesProducerAndConsumer() throws Exception {
