diff --git a/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java b/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java index 0d800b3c981..71aabf6e3b0 100644 --- a/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java +++ b/fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java @@ -201,9 +201,9 @@ public void close() throws Exception { } if (lookupClient != null) { - // timeout is Long.MAX_VALUE to make the pending get request - // to be processed - lookupClient.close(Duration.ofMillis(Long.MAX_VALUE)); + // use the configured close timeout; by default it is Long.MAX_VALUE to let + // pending lookup requests be processed before close + lookupClient.close(conf.get(ConfigOptions.CLIENT_LOOKUP_CLOSE_TIMEOUT)); } if (remoteFileDownloader != null) { diff --git a/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupClientCloseTest.java b/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupClientCloseTest.java new file mode 100644 index 00000000000..cf51fff02f4 --- /dev/null +++ b/fluss-client/src/test/java/org/apache/fluss/client/lookup/LookupClientCloseTest.java @@ -0,0 +1,93 @@ +/* + * 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.fluss.client.lookup; + +import org.apache.fluss.client.metadata.TestingMetadataUpdater; +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; + +/** Tests for {@link LookupClient} close behavior with configurable timeout. */ +class LookupClientCloseTest { + + @Test + void testCloseWithVeryShortTimeoutReturnsNormally() { + // Create a lookup client with default config (no pending lookups) + LookupClient lookupClient = createLookupClient(); + + // Closing with a 1ms timeout should return normally without throwing + // InterruptedException, even though there are no pending requests to wait for. + assertThatCode(() -> lookupClient.close(Duration.ofMillis(1))).doesNotThrowAnyException(); + } + + @Test + void testCloseWithZeroTimeoutReturnsNormally() { + LookupClient lookupClient = createLookupClient(); + + // A zero-duration close should also return without error. + assertThatCode(() -> lookupClient.close(Duration.ZERO)).doesNotThrowAnyException(); + } + + @Test + void testCloseWithLargeTimeoutReturnsNormally() { + LookupClient lookupClient = createLookupClient(); + + // Closing with a large timeout should also work fine when there are no pending requests. + assertThatCode(() -> lookupClient.close(Duration.ofSeconds(5))).doesNotThrowAnyException(); + } + + @Test + void testCloseIsIdempotent() { + LookupClient lookupClient = createLookupClient(); + + // First close should succeed + lookupClient.close(Duration.ofMillis(100)); + + // Second close should also return without throwing + assertThatCode(() -> lookupClient.close(Duration.ofMillis(100))).doesNotThrowAnyException(); + } + + @Test + void testDefaultCloseTimeoutConfigValue() { + // Verify the default value of CLIENT_LOOKUP_CLOSE_TIMEOUT is Long.MAX_VALUE + // which preserves the previous behavior of waiting indefinitely. + Configuration conf = new Configuration(); + assertThat(conf.get(ConfigOptions.CLIENT_LOOKUP_CLOSE_TIMEOUT)) + .isEqualTo(Duration.ofMillis(Long.MAX_VALUE)); + } + + private LookupClient createLookupClient() { + Map tableInfos = new HashMap<>(); + TestingMetadataUpdater metadataUpdater = TestingMetadataUpdater.builder(tableInfos).build(); + Configuration conf = new Configuration(); + // Use small queue to avoid resource waste + conf.set(ConfigOptions.CLIENT_LOOKUP_QUEUE_SIZE, 5); + conf.set(ConfigOptions.CLIENT_LOOKUP_MAX_BATCH_SIZE, 10); + return new LookupClient(conf, metadataUpdater); + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index 233902cb391..5f1bbf2ba3e 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -1518,6 +1518,17 @@ public class ConfigOptions { "Setting a value greater than zero will cause the client to resend any lookup request " + "that fails with a potentially transient error."); + public static final ConfigOption CLIENT_LOOKUP_CLOSE_TIMEOUT = + key("client.lookup.close-timeout") + .durationType() + .defaultValue(Duration.ofMillis(Long.MAX_VALUE)) + .withDescription( + "The timeout for closing the lookup client. By default, the lookup client " + + "will wait indefinitely for pending lookup requests to complete during " + + "close. Set a smaller value to bound the close time, which is useful " + + "in scenarios where the client must shut down within a bounded time " + + "(e.g., on SIGTERM in Kubernetes)."); + public static final ConfigOption CLIENT_SCANNER_REMOTE_LOG_PREFETCH_NUM = key("client.scanner.remote-log.prefetch-num") .intType()