This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 4c4b81859b [Fix][Connector-V2] Wait for Kudu client shutdown (#11585)
4c4b81859b is described below
commit 4c4b81859b090f9e8c724970c6995952de24bb4d
Author: zhiwei.niu <[email protected]>
AuthorDate: Wed Jul 29 23:51:59 2026 +0800
[Fix][Connector-V2] Wait for Kudu client shutdown (#11585)
---
.../seatunnel/kudu/catalog/KuduCatalog.java | 14 +++-
.../kudu/kuduclient/KuduClientResource.java | 82 ++++++++++++++++++++++
.../seatunnel/kudu/kuduclient/KuduInputFormat.java | 12 +++-
.../kudu/kuduclient/KuduOutputFormat.java | 32 +++++++--
.../seatunnel/kudu/source/KuduSourceReader.java | 22 +++---
.../kudu/source/KuduSourceSplitEnumerator.java | 35 +++++----
.../connectors/seatunnel/kudu/util/KuduUtil.java | 65 ++++++++++++++---
.../kudu/kuduclient/KuduClientResourceTest.java | 82 ++++++++++++++++++++++
8 files changed, 304 insertions(+), 40 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/catalog/KuduCatalog.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/catalog/KuduCatalog.java
index e03e223463..3d24cd480f 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/catalog/KuduCatalog.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/catalog/KuduCatalog.java
@@ -34,6 +34,7 @@ import
org.apache.seatunnel.api.table.catalog.exception.TableNotExistException;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.connectors.seatunnel.kudu.config.CommonConfig;
import org.apache.seatunnel.connectors.seatunnel.kudu.config.KuduBaseOptions;
+import
org.apache.seatunnel.connectors.seatunnel.kudu.kuduclient.KuduClientResource;
import
org.apache.seatunnel.connectors.seatunnel.kudu.kuduclient.KuduTypeMapper;
import org.apache.seatunnel.connectors.seatunnel.kudu.util.KuduUtil;
@@ -59,6 +60,7 @@ public class KuduCatalog implements Catalog {
private final CommonConfig config;
private KuduClient kuduClient;
+ private KuduClientResource kuduClientResource;
private final String defaultDatabase = "default_database";
@@ -71,15 +73,23 @@ public class KuduCatalog implements Catalog {
@Override
public void open() throws CatalogException {
- kuduClient = KuduUtil.getKuduClient(config);
+ kuduClientResource = KuduUtil.getKuduClientResource(config);
+ kuduClient = kuduClientResource.getClient();
}
@Override
public void close() throws CatalogException {
try {
- kuduClient.close();
+ if (kuduClientResource != null) {
+ kuduClientResource.close();
+ } else if (kuduClient != null) {
+ kuduClient.close();
+ }
} catch (KuduException e) {
throw new CatalogException("Failed close kudu client", e);
+ } finally {
+ kuduClient = null;
+ kuduClientResource = null;
}
}
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResource.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResource.java
new file mode 100644
index 0000000000..0b07e8fc0f
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResource.java
@@ -0,0 +1,82 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.kudu.kuduclient;
+
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduException;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Owns a Kudu client and the executor used by its Netty event loops.
+ *
+ * <p>Kudu 1.15 initiates a graceful Netty shutdown from {@link
KuduClient#close()} but returns
+ * before the event-loop threads terminate. Waiting for the supplied executor
prevents those threads
+ * from accessing connector classes after an engine unloads the user-code
classloader.
+ */
+@Slf4j
+public class KuduClientResource implements AutoCloseable {
+
+ private static final long GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS = 20L;
+ private static final long FORCED_SHUTDOWN_TIMEOUT_SECONDS = 5L;
+
+ private final KuduClient kuduClient;
+ private final ExecutorService executorService;
+
+ public KuduClientResource(KuduClient kuduClient, ExecutorService
executorService) {
+ this.kuduClient = kuduClient;
+ this.executorService = executorService;
+ }
+
+ public KuduClient getClient() {
+ return kuduClient;
+ }
+
+ @Override
+ public void close() throws KuduException {
+ try {
+ kuduClient.close();
+ } finally {
+ awaitExecutorTermination();
+ }
+ }
+
+ private void awaitExecutorTermination() {
+ executorService.shutdown();
+ try {
+ if (executorService.awaitTermination(
+ GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ return;
+ }
+
+ log.warn("Kudu client executor did not terminate gracefully,
forcing shutdown.");
+ executorService.shutdownNow();
+ if (!executorService.awaitTermination(
+ FORCED_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ log.warn("Kudu client executor did not terminate after forced
shutdown.");
+ }
+ } catch (InterruptedException e) {
+ executorService.shutdownNow();
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted while waiting for Kudu client executor to
terminate.", e);
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduInputFormat.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduInputFormat.java
index b34cda3dd9..90a41d85e2 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduInputFormat.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduInputFormat.java
@@ -56,13 +56,16 @@ public class KuduInputFormat implements Serializable {
/** Declare the global variable KuduClient and use it to manipulate the
Kudu table */
public KuduClient kuduClient;
+ private transient KuduClientResource kuduClientResource;
+
public KuduInputFormat(@NonNull KuduSourceConfig kuduSourceConfig) {
this.kuduSourceConfig = kuduSourceConfig;
}
public void openInputFormat() {
if (kuduClient == null) {
- kuduClient = KuduUtil.getKuduClient(kuduSourceConfig);
+ kuduClientResource =
KuduUtil.getKuduClientResource(kuduSourceConfig);
+ kuduClient = kuduClientResource.getClient();
}
}
@@ -84,12 +87,17 @@ public class KuduInputFormat implements Serializable {
public void closeInputFormat() {
if (kuduClient != null) {
try {
- kuduClient.close();
+ if (kuduClientResource != null) {
+ kuduClientResource.close();
+ } else {
+ kuduClient.close();
+ }
} catch (KuduException e) {
throw new KuduConnectorException(
KuduConnectorErrorCode.CLOSE_KUDU_CLIENT_FAILED, e);
} finally {
kuduClient = null;
+ kuduClientResource = null;
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduOutputFormat.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduOutputFormat.java
index 01667c6e7d..4e9978291a 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduOutputFormat.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduOutputFormat.java
@@ -51,6 +51,7 @@ public class KuduOutputFormat implements Serializable {
private final KuduSinkConfig.SaveMode saveMode;
private final KuduSinkConfig kuduSinkConfig;
private KuduClient kuduClient;
+ private transient KuduClientResource kuduClientResource;
private KuduSession kuduSession;
private KuduTable kuduTable;
@@ -71,18 +72,36 @@ public class KuduOutputFormat implements Serializable {
}
private void openOutputFormat() {
- this.kuduClient = KuduUtil.getKuduClient(kuduSinkConfig);
- this.kuduSession = getSession();
try {
+ this.kuduClientResource =
KuduUtil.getKuduClientResource(kuduSinkConfig);
+ this.kuduClient = kuduClientResource.getClient();
+ this.kuduSession = getSession();
kuduTable = kuduClient.openTable(kuduTableName);
+ seaTunnelRowSerializer = new KuduRowSerializer(kuduTable,
saveMode, seaTunnelRowType);
} catch (KuduException e) {
+ closeClientAfterInitializationFailure(e);
throw new
KuduConnectorException(KuduConnectorErrorCode.INIT_KUDU_CLIENT_FAILED, e);
+ } catch (RuntimeException e) {
+ closeClientAfterInitializationFailure(e);
+ throw e;
}
log.info(
"The Kudu client for Master: {} is initialized successfully.",
kuduSinkConfig.getMasters());
+ }
- seaTunnelRowSerializer = new KuduRowSerializer(kuduTable, saveMode,
seaTunnelRowType);
+ private void closeClientAfterInitializationFailure(Exception
initializationException) {
+ if (kuduClientResource == null) {
+ return;
+ }
+ try {
+ kuduClientResource.close();
+ } catch (Exception closeException) {
+ initializationException.addSuppressed(closeException);
+ } finally {
+ kuduClient = null;
+ kuduClientResource = null;
+ }
}
private KuduSession getSession() {
@@ -108,11 +127,16 @@ public class KuduOutputFormat implements Serializable {
log.error("Error while closing session.", e);
}
try {
- if (kuduClient != null) {
+ if (kuduClientResource != null) {
+ kuduClientResource.close();
+ } else if (kuduClient != null) {
kuduClient.close();
}
} catch (Exception e) {
log.error("Error while closing client.", e);
+ } finally {
+ kuduClient = null;
+ kuduClientResource = null;
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceReader.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceReader.java
index 5a83d55429..5871b9179b 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceReader.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceReader.java
@@ -84,14 +84,20 @@ public class KuduSourceReader implements
SourceReader<SeaTunnelRow, KuduSourceSp
TablePath tablePath = split.getTablePath();
SeaTunnelRowType seaTunnelRowType = tables.get(tablePath);
KuduScanner kuduScanner =
kuduInputFormat.scanner(split.getToken());
- while (kuduScanner.hasMoreRows()) {
- RowResultIterator rowResults = kuduScanner.nextRows();
- while (rowResults.hasNext()) {
- RowResult rowResult = rowResults.next();
- SeaTunnelRow seaTunnelRow =
- kuduInputFormat.toInternal(rowResult,
seaTunnelRowType);
- seaTunnelRow.setTableId(tablePath.toString());
- output.collect(seaTunnelRow);
+ try {
+ while (kuduScanner.hasMoreRows()) {
+ RowResultIterator rowResults = kuduScanner.nextRows();
+ while (rowResults.hasNext()) {
+ RowResult rowResult = rowResults.next();
+ SeaTunnelRow seaTunnelRow =
+ kuduInputFormat.toInternal(rowResult,
seaTunnelRowType);
+ seaTunnelRow.setTableId(tablePath.toString());
+ output.collect(seaTunnelRow);
+ }
+ }
+ } finally {
+ if (!kuduScanner.isClosed()) {
+ kuduScanner.close();
}
}
} else if (noMoreSplit && splits.isEmpty()) {
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceSplitEnumerator.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceSplitEnumerator.java
index 6121e005a4..0d23c99beb 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceSplitEnumerator.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/source/KuduSourceSplitEnumerator.java
@@ -93,26 +93,33 @@ public class KuduSourceSplitEnumerator
@Override
public void run() throws IOException {
+ try {
+ Set<Integer> readers = enumeratorContext.registeredReaders();
+ while (!pendingTables.isEmpty()) {
+ synchronized (stateLock) {
+ TablePath tablePath = pendingTables.poll();
+ log.info("Splitting table {}.", tablePath);
- Set<Integer> readers = enumeratorContext.registeredReaders();
- while (!pendingTables.isEmpty()) {
- synchronized (stateLock) {
- TablePath tablePath = pendingTables.poll();
- log.info("Splitting table {}.", tablePath);
+ Collection<KuduSourceSplit> splits =
discoverySplits(tables.get(tablePath));
+ log.info("Split table {} into {} splits.", tablePath,
splits.size());
- Collection<KuduSourceSplit> splits =
discoverySplits(tables.get(tablePath));
- log.info("Split table {} into {} splits.", tablePath,
splits.size());
+ addPendingSplit(splits);
+ }
- addPendingSplit(splits);
+ synchronized (stateLock) {
+ assignSplit(readers);
+ }
}
- synchronized (stateLock) {
- assignSplit(readers);
- }
+ log.info(
+ "No more splits to assign." + " Sending NoMoreSplitsEvent
to reader {}.",
+ readers);
+ readers.forEach(enumeratorContext::signalNoMoreSplits);
+ } finally {
+ // Scan tokens contain everything readers need, so the enumerator
client can be
+ // released immediately instead of retaining Netty threads until
job teardown.
+ kuduInputFormat.closeInputFormat();
}
-
- log.info("No more splits to assign." + " Sending NoMoreSplitsEvent to
reader {}.", readers);
- readers.forEach(enumeratorContext::signalNoMoreSplits);
}
private Set<KuduSourceSplit> discoverySplits(KuduSourceTableConfig
kuduSourceTableConfig)
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
index 1c7989f4da..3faded4a5e 100644
---
a/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
+++
b/seatunnel-connectors-v2/connector-kudu/src/main/java/org/apache/seatunnel/connectors/seatunnel/kudu/util/KuduUtil.java
@@ -25,6 +25,7 @@ import
org.apache.seatunnel.connectors.seatunnel.kudu.config.KuduSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.kudu.config.KuduSourceTableConfig;
import
org.apache.seatunnel.connectors.seatunnel.kudu.exception.KuduConnectorErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.kudu.exception.KuduConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.kudu.kuduclient.KuduClientResource;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
@@ -46,6 +47,9 @@ import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -68,13 +72,50 @@ public class KuduUtil {
UserGroupInformation ugi = loginAndReturnUgi(config);
return ugi.doAs(
(PrivilegedExceptionAction<KuduClient>)
- () -> getKuduClientInternal(config));
+ () -> getKuduClientInternal(config, null));
}
}
- return getKuduClientInternal(config);
+ return getKuduClientInternal(config, null);
+ } catch (IOException | InterruptedException e) {
+ throw new
KuduConnectorException(KuduConnectorErrorCode.INIT_KUDU_CLIENT_FAILED, e);
+ }
+ }
+ public static KuduClientResource getKuduClientResource(CommonConfig
config) {
+ AtomicInteger threadCounter = new AtomicInteger();
+ ExecutorService executorService =
+ Executors.newCachedThreadPool(
+ runnable -> {
+ Thread thread =
+ new Thread(
+ runnable,
+ "seatunnel-kudu-nio-"
+ +
threadCounter.incrementAndGet());
+ thread.setDaemon(true);
+ return thread;
+ });
+ boolean initialized = false;
+ try {
+ KuduClient kuduClient;
+ if (config.getEnableKerberos()) {
+ synchronized (UserGroupInformation.class) {
+ UserGroupInformation ugi = loginAndReturnUgi(config);
+ kuduClient =
+ ugi.doAs(
+ (PrivilegedExceptionAction<KuduClient>)
+ () ->
getKuduClientInternal(config, executorService));
+ }
+ } else {
+ kuduClient = getKuduClientInternal(config, executorService);
+ }
+ initialized = true;
+ return new KuduClientResource(kuduClient, executorService);
} catch (IOException | InterruptedException e) {
throw new
KuduConnectorException(KuduConnectorErrorCode.INIT_KUDU_CLIENT_FAILED, e);
+ } finally {
+ if (!initialized) {
+ executorService.shutdownNow();
+ }
}
}
@@ -109,14 +150,18 @@ public class KuduUtil {
}
}
- private static KuduClient getKuduClientInternal(CommonConfig config) {
- return new AsyncKuduClient.AsyncKuduClientBuilder(
- Arrays.asList(config.getMasters().split(",")))
- .workerCount(config.getWorkerCount())
-
.defaultAdminOperationTimeoutMs(config.getAdminOperationTimeout())
- .defaultOperationTimeoutMs(config.getOperationTimeout())
- .build()
- .syncClient();
+ private static KuduClient getKuduClientInternal(
+ CommonConfig config, ExecutorService executorService) {
+ AsyncKuduClient.AsyncKuduClientBuilder builder =
+ new AsyncKuduClient.AsyncKuduClientBuilder(
+ Arrays.asList(config.getMasters().split(",")))
+ .workerCount(config.getWorkerCount())
+
.defaultAdminOperationTimeoutMs(config.getAdminOperationTimeout())
+
.defaultOperationTimeoutMs(config.getOperationTimeout());
+ if (executorService != null) {
+ builder.nioExecutor(executorService);
+ }
+ return builder.build().syncClient();
}
public static List<KuduScanToken> getKuduScanToken(
diff --git
a/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
b/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
new file mode 100644
index 0000000000..4b2cb2ad4e
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kudu/src/test/java/org/apache/seatunnel/connectors/seatunnel/kudu/kuduclient/KuduClientResourceTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.kudu.kuduclient;
+
+import org.apache.kudu.client.KuduClient;
+import org.apache.kudu.client.KuduException;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import org.mockito.Mockito;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+class KuduClientResourceTest {
+
+ @Test
+ void shouldWaitForExecutorAfterClosingClient() throws Exception {
+ KuduClient kuduClient = Mockito.mock(KuduClient.class);
+ ExecutorService executorService = Mockito.mock(ExecutorService.class);
+ Mockito.when(executorService.awaitTermination(20L,
TimeUnit.SECONDS)).thenReturn(true);
+
+ new KuduClientResource(kuduClient, executorService).close();
+
+ InOrder inOrder = Mockito.inOrder(kuduClient, executorService);
+ inOrder.verify(kuduClient).close();
+ inOrder.verify(executorService).shutdown();
+ inOrder.verify(executorService).awaitTermination(20L,
TimeUnit.SECONDS);
+ Mockito.verify(executorService, Mockito.never()).shutdownNow();
+ }
+
+ @Test
+ void shouldForceShutdownWhenGracefulShutdownTimesOut() throws Exception {
+ KuduClient kuduClient = Mockito.mock(KuduClient.class);
+ ExecutorService executorService = Mockito.mock(ExecutorService.class);
+ Mockito.when(executorService.awaitTermination(20L,
TimeUnit.SECONDS)).thenReturn(false);
+ Mockito.when(executorService.awaitTermination(5L,
TimeUnit.SECONDS)).thenReturn(true);
+
+ new KuduClientResource(kuduClient, executorService).close();
+
+ InOrder inOrder = Mockito.inOrder(kuduClient, executorService);
+ inOrder.verify(kuduClient).close();
+ inOrder.verify(executorService).shutdown();
+ inOrder.verify(executorService).awaitTermination(20L,
TimeUnit.SECONDS);
+ inOrder.verify(executorService).shutdownNow();
+ inOrder.verify(executorService).awaitTermination(5L, TimeUnit.SECONDS);
+ }
+
+ @Test
+ void shouldShutdownExecutorWhenClosingClientFails() throws Exception {
+ KuduClient kuduClient = Mockito.mock(KuduClient.class);
+ KuduException closeException = Mockito.mock(KuduException.class);
+ Mockito.doThrow(closeException).when(kuduClient).close();
+ ExecutorService executorService = Mockito.mock(ExecutorService.class);
+ Mockito.when(executorService.awaitTermination(20L,
TimeUnit.SECONDS)).thenReturn(true);
+
+ KuduException actualException =
+ Assertions.assertThrows(
+ KuduException.class,
+ () -> new KuduClientResource(kuduClient,
executorService).close());
+
+ Assertions.assertSame(closeException, actualException);
+ Mockito.verify(executorService).shutdown();
+ Mockito.verify(executorService).awaitTermination(20L,
TimeUnit.SECONDS);
+ }
+}
