This is an automated email from the ASF dual-hosted git repository.
nsivabalan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 3483d602d3b6 fix(client): do not block the writer on a running async
table service when switching write clients (#20444)
3483d602d3b6 is described below
commit 3483d602d3b6b24abca0d8fa8fdabcddaa63d3bc
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Wed Oct 28 19:46:01 2026 -0700
fix(client): do not block the writer on a running async table service when
switching write clients (#20444)
AsyncCompactService and AsyncClusteringService held the same monitor for a
whole compaction/clustering run and for updateWriteClient, so switching the
write client waited for the running run to complete. The caller is the
writer's driver thread (HoodieStreamer on write client re-creation, the
Spark
streaming sink inside addBatch on a schema change), and a compaction or
clustering can run for an hour or more, e.g. while waiting for cluster
resources; ingestion must not stall for that long.
Restore the non-blocking switch: later runs use the new client, and a
running
run keeps the client it started with. Make the switched client fields
volatile, as the service thread reads them while the writer switches them.
---
.../apache/hudi/async/AsyncClusteringService.java | 13 ++--
.../org/apache/hudi/async/AsyncCompactService.java | 13 ++--
.../java/org/apache/hudi/async/BaseClusterer.java | 3 +-
.../java/org/apache/hudi/async/BaseCompactor.java | 3 +-
.../async/TestAsyncTableServiceClientHandoff.java | 69 +++++++++-------------
.../org/apache/hudi/HoodieStreamingSink.scala | 5 +-
6 files changed, 44 insertions(+), 62 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
index 236a6c10e964..a1be465977ac 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncClusteringService.java
@@ -76,9 +76,7 @@ public abstract class AsyncClusteringService extends
HoodieAsyncTableService {
final String instant = fetchNextAsyncServiceInstant();
if (null != instant) {
log.info("Starting clustering for instant {}", instant);
- synchronized (clusteringClient) {
- clusteringClient.cluster(instant);
- }
+ clusteringClient.cluster(instant);
log.info("Finished clustering for instant {}", instant);
}
}
@@ -100,11 +98,10 @@ public abstract class AsyncClusteringService extends
HoodieAsyncTableService {
}
/**
- * Switches clients between complete clustering runs. Once this returns, no
run uses the old client.
+ * Switches the client used by later clustering runs. Does not wait for a
clustering that is running: it can run
+ * for a long time, and the writer calling this must not be blocked until it
completes.
*/
- public void updateWriteClient(BaseHoodieWriteClient writeClient) {
- synchronized (clusteringClient) {
- clusteringClient.updateWriteClient(writeClient);
- }
+ public synchronized void updateWriteClient(BaseHoodieWriteClient
writeClient) {
+ this.clusteringClient.updateWriteClient(writeClient);
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
index 6fb6514e56ec..72e3acd7dbef 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/AsyncCompactService.java
@@ -78,9 +78,7 @@ public abstract class AsyncCompactService extends
HoodieAsyncTableService {
if (null != instantTime) {
log.info("Starting Compaction for instant {}", instantTime);
- synchronized (compactor) {
- compactor.compact(instantTime);
- }
+ compactor.compact(instantTime);
log.info("Finished Compaction for instant {}", instantTime);
}
}
@@ -111,11 +109,10 @@ public abstract class AsyncCompactService extends
HoodieAsyncTableService {
}
/**
- * Switches clients between complete compaction runs. Once this returns, no
run uses the old client.
+ * Switches the client used by later compaction runs. Does not wait for a
compaction that is running: it can run
+ * for a long time, and the writer calling this must not be blocked until it
completes.
*/
- public void updateWriteClient(BaseHoodieWriteClient writeClient) {
- synchronized (compactor) {
- compactor.updateWriteClient(writeClient);
- }
+ public synchronized void updateWriteClient(BaseHoodieWriteClient
writeClient) {
+ this.compactor.updateWriteClient(writeClient);
}
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseClusterer.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseClusterer.java
index bde48d95b21d..47704d667835 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseClusterer.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseClusterer.java
@@ -31,7 +31,8 @@ public abstract class BaseClusterer<T, I, K, O> implements
Serializable {
private static final long serialVersionUID = 1L;
- protected transient BaseHoodieWriteClient<T, I, K, O> clusteringClient;
+ // volatile: switched by the writer while the async service thread reads it
at the start of each run
+ protected transient volatile BaseHoodieWriteClient<T, I, K, O>
clusteringClient;
public BaseClusterer(BaseHoodieWriteClient<T, I, K, O> clusteringClient) {
this.clusteringClient = clusteringClient;
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseCompactor.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseCompactor.java
index e8ed9976f191..5ca18508eaa5 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseCompactor.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/BaseCompactor.java
@@ -30,7 +30,8 @@ public abstract class BaseCompactor<T, I, K, O> implements
Serializable {
private static final long serialVersionUID = 1L;
- protected transient BaseHoodieWriteClient<T, I, K, O> compactionClient;
+ // volatile: switched by the writer while the async service thread reads it
at the start of each run
+ protected transient volatile BaseHoodieWriteClient<T, I, K, O>
compactionClient;
public BaseCompactor(BaseHoodieWriteClient<T, I, K, O> compactionClient) {
this.compactionClient = compactionClient;
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
index 3d8a8543ae7c..8f00e69c5e68 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceClientHandoff.java
@@ -24,35 +24,35 @@ import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.config.HoodieWriteConfig;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.stubbing.Answer;
-import java.io.IOException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class TestAsyncTableServiceClientHandoff {
+ /**
+ * A compaction or clustering run can take a long time (e.g. waiting for
cluster resources), so switching the
+ * client must not block the writer until the run completes. The running run
keeps the client it started with,
+ * and the next run uses the new client.
+ */
@ParameterizedTest
- @CsvSource({"true,false", "false,false", "true,true", "false,true"})
- void testHandoffWaitsForActiveRun(boolean compact, boolean failRun) throws
Exception {
+ @ValueSource(booleans = {true, false})
+ void testHandoffDoesNotWaitForActiveRun(boolean compact) throws Exception {
HoodieEngineContext context = mock(HoodieEngineContext.class);
BaseHoodieWriteClient oldClient = mock(BaseHoodieWriteClient.class);
BaseHoodieWriteClient newClient = mock(BaseHoodieWriteClient.class);
@@ -60,22 +60,20 @@ class TestAsyncTableServiceClientHandoff {
when(oldClient.getConfig()).thenReturn(config);
when(config.areTableServicesEnabled()).thenReturn(true);
AtomicReference<BaseHoodieWriteClient> currentClient = new
AtomicReference<>(oldClient);
+ AtomicReference<BaseHoodieWriteClient> firstRunClient = new
AtomicReference<>();
+ AtomicReference<BaseHoodieWriteClient> secondRunClient = new
AtomicReference<>();
CountDownLatch entered = new CountDownLatch(1);
CountDownLatch finishRun = new CountDownLatch(1);
+ CountDownLatch firstRunDone = new CountDownLatch(1);
CountDownLatch nextRun = new CountDownLatch(1);
- CountDownLatch stopped = new CountDownLatch(1);
Answer<Void> run = invocation -> {
- BaseHoodieWriteClient capturedClient = currentClient.get();
if ("001".equals(invocation.getArgument(0))) {
+ firstRunClient.set(currentClient.get());
entered.countDown();
- assertTrue(finishRun.await(10, TimeUnit.SECONDS));
- assertSame(oldClient, capturedClient);
- verify(oldClient, never()).close();
- if (failRun) {
- throw new IOException("Injected async operation failure");
- }
+ assertTrue(finishRun.await(30, TimeUnit.SECONDS));
+ firstRunDone.countDown();
} else {
- assertSame(newClient, capturedClient);
+ secondRunClient.set(currentClient.get());
nextRun.countDown();
}
return null;
@@ -114,34 +112,21 @@ class TestAsyncTableServiceClientHandoff {
ExecutorService swapExecutor = Executors.newSingleThreadExecutor();
try {
- service.start(errored -> {
- stopped.countDown();
- return true;
- });
+ service.start(errored -> true);
service.enqueuePendingAsyncServiceInstant("001");
assertTrue(entered.await(10, TimeUnit.SECONDS));
- CountDownLatch swapping = new CountDownLatch(1);
- Future<?> swap = swapExecutor.submit(() -> {
- swapping.countDown();
- handoff.accept(newClient);
- oldClient.close();
- });
- assertTrue(swapping.await(10, TimeUnit.SECONDS));
- assertThrows(TimeoutException.class, () -> swap.get(200,
TimeUnit.MILLISECONDS));
- verify(oldClient, never()).close();
+
+ // the handoff returns while the run is still active
+ swapExecutor.submit(() -> handoff.accept(newClient)).get(10,
TimeUnit.SECONDS);
+ assertFalse(firstRunDone.await(0, TimeUnit.SECONDS), "the run must still
be active after the handoff");
+ assertSame(oldClient, firstRunClient.get());
finishRun.countDown();
- swap.get(10, TimeUnit.SECONDS);
- verify(oldClient).close();
- if (failRun) {
- assertTrue(stopped.await(10, TimeUnit.SECONDS));
- assertTrue(service.hasError());
- } else {
- service.enqueuePendingAsyncServiceInstant("002");
- assertTrue(nextRun.await(10, TimeUnit.SECONDS));
- // An idle service must hand back the client promptly too, without
waiting for another run.
- swapExecutor.submit(() -> handoff.accept(oldClient)).get(10,
TimeUnit.SECONDS);
- }
+ assertTrue(firstRunDone.await(10, TimeUnit.SECONDS));
+ service.enqueuePendingAsyncServiceInstant("002");
+ assertTrue(nextRun.await(10, TimeUnit.SECONDS));
+ assertSame(newClient, secondRunClient.get());
+ assertFalse(service.hasError());
} finally {
finishRun.countDown();
service.shutdown(true);
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
index 7410bbf29011..4f846ef48bb8 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala
@@ -315,8 +315,9 @@ class HoodieStreamingSink(sqlContext: SQLContext,
}
/**
- * Each handoff waits for the service's current run before switching to the
new client. Close the previous
- * client only after both services have handed it back, without retaining
clients until query termination.
+ * Same as HoodieStreamer on a schema change (StreamSync#reInitWriteClient
and
+ * HoodieStreamer#onInitializingWriteClient): the async table services
switch to the new client for their next
+ * runs, and the previous client is closed right away.
*/
private def retireWriteClient(previousClient: SparkRDDWriteClient[_],
newClient: SparkRDDWriteClient[_]): Unit = {
if (asyncCompactorService != null) {
