This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 54347de81f [flink] Shut down KvQueryServer when QueryExecutorOperator
closes (#9287)
54347de81f is described below
commit 54347de81fb0758a0425af40573acfd6fde64a54
Author: Eunbin Son <[email protected]>
AuthorDate: Mon Aug 24 00:27:24 2026 +0900
[flink] Shut down KvQueryServer when QueryExecutorOperator closes (#9287)
---
.../flink/service/QueryExecutorOperator.java | 8 +-
.../flink/service/QueryExecutorOperatorTest.java | 97 ++++++++++++++++++++++
2 files changed, 104 insertions(+), 1 deletion(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
index cd81dfa35a..ed4758e211 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryExecutorOperator.java
@@ -60,6 +60,8 @@ public class QueryExecutorOperator extends
AbstractStreamOperator<InternalRow>
private transient IOManager ioManager;
+ private transient KvQueryServer server;
+
public QueryExecutorOperator(Table table) {
this.table = table;
}
@@ -86,7 +88,7 @@ public class QueryExecutorOperator extends
AbstractStreamOperator<InternalRow>
.newLocalTableQuery()
.withIOManager(ioManager)
.withMetrics(metrics);
- KvQueryServer server =
+ this.server =
new KvQueryServer(
RuntimeContextUtils.getIndexOfThisSubtask(getRuntimeContext()),
RuntimeContextUtils.getNumberOfParallelSubtasks(getRuntimeContext()),
@@ -128,6 +130,10 @@ public class QueryExecutorOperator extends
AbstractStreamOperator<InternalRow>
@Override
public void close() throws Exception {
super.close();
+ // shut down the server first, so that no in-flight request can hit
the closed query
+ if (server != null) {
+ server.shutdown();
+ }
if (query != null) {
query.close();
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/service/QueryExecutorOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/service/QueryExecutorOperatorTest.java
new file mode 100644
index 0000000000..9c55962086
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/service/QueryExecutorOperatorTest.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.service;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.types.DataTypes;
+
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link QueryExecutorOperator}. */
+public class QueryExecutorOperatorTest extends TableTestBase {
+
+ private static final int CONNECT_TIMEOUT_MS = 10_000;
+
+ @Test
+ public void testQueryServerIsShutDownOnClose() throws Exception {
+ Identifier identifier = identifier("query_executor_table");
+ Schema schema =
+ Schema.newBuilder()
+ .column("k", DataTypes.INT())
+ .column("v", DataTypes.INT())
+ .primaryKey("k")
+ .option(CoreOptions.BUCKET.key(), "1")
+ .build();
+ catalog.createTable(identifier, schema, false);
+ FileStoreTable table = getTable(identifier);
+
+ OneInputStreamOperatorTestHarness<InternalRow, InternalRow> harness =
+ new OneInputStreamOperatorTestHarness<>(new
QueryExecutorOperator(table));
+ harness.setup();
+ harness.initializeEmptyState();
+ harness.open();
+
+ InetSocketAddress address = serverAddress(harness);
+ assertThatCode(() -> connect(address)).doesNotThrowAnyException();
+
+ harness.close();
+
+ assertThatThrownBy(() ->
connect(address)).isInstanceOf(ConnectException.class);
+ }
+
+ /**
+ * Reads the address the operator has published downstream. The output row
is built by {@link
+ * QueryExecutorOperator#outputType()}: parallelism, subtask index, host,
port.
+ */
+ @SuppressWarnings("unchecked")
+ private static InetSocketAddress serverAddress(
+ OneInputStreamOperatorTestHarness<InternalRow, InternalRow>
harness) {
+ List<InternalRow> rows = new ArrayList<>();
+ for (Object record : harness.getOutput()) {
+ rows.add(((StreamRecord<InternalRow>) record).getValue());
+ }
+ assertThat(rows).hasSize(1);
+ InternalRow row = rows.get(0);
+ return new InetSocketAddress(row.getString(2).toString(),
row.getInt(3));
+ }
+
+ private static void connect(InetSocketAddress address) throws IOException {
+ try (Socket socket = new Socket()) {
+ socket.connect(address, CONNECT_TIMEOUT_MS);
+ }
+ }
+}
