This is an automated email from the ASF dual-hosted git repository.
danny0405 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 302040ed46ea fix(utilities): surface async table service failures in
HoodieStreamer (#20429)
302040ed46ea is described below
commit 302040ed46ea4e9af3366d76d02dba96101507ab
Author: Sivabalan Narayanan <[email protected]>
AuthorDate: Tue Nov 3 18:16:37 2026 -0800
fix(utilities): surface async table service failures in HoodieStreamer
(#20429)
- Record the cause when async compaction or clustering stops with an error,
expose it via HoodieAsyncService#getFailure(), and clear it on restart.
- Check for a failed async table service on every continuous ingestion
round, not only when scheduling new work, and fail with the original
cause attached.
- In StreamSync#reInitWriteClient, hand the new write client to the async
table services before closing the previous one, and always close the
previous client.
Co-authored-by: danny0405 <[email protected]>
---
.../apache/hudi/async/AsyncClusteringService.java | 6 +-
.../org/apache/hudi/async/AsyncCompactService.java | 6 +-
.../org/apache/hudi/async/HoodieAsyncService.java | 27 +++-
.../hudi/async/TestAsyncTableServiceFailure.java | 142 +++++++++++++++++++++
.../hudi/async/TestHoodieAsyncTableService.java | 56 ++++++++
.../hudi/utilities/streamer/HoodieStreamer.java | 27 ++--
.../apache/hudi/utilities/streamer/StreamSync.java | 16 ++-
...TestHoodieStreamerAsyncTableServiceFailure.java | 131 +++++++++++++++++++
8 files changed, 390 insertions(+), 21 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 a1be465977ac..e0b0637c4e44 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
@@ -82,14 +82,14 @@ public abstract class AsyncClusteringService extends
HoodieAsyncTableService {
}
log.info("Clustering executor shutting down properly");
} catch (InterruptedException ie) {
- hasError = true;
+ recordFailure(ie);
log.warn("Clustering executor got interrupted exception! Stopping",
ie);
} catch (IOException e) {
- hasError = true;
+ recordFailure(e);
log.error("Clustering executor failed due to IOException", e);
throw new HoodieIOException(e.getMessage(), e);
} catch (Exception e) {
- hasError = true;
+ recordFailure(e);
log.error("Clustering executor failed", e);
throw e;
}
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 72e3acd7dbef..e092aa32455d 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
@@ -84,14 +84,14 @@ public abstract class AsyncCompactService extends
HoodieAsyncTableService {
}
log.info("Compactor shutting down properly!!");
} catch (InterruptedException ie) {
- hasError = true;
+ recordFailure(ie);
log.warn("Compactor executor thread got interrupted exception.
Stopping", ie);
} catch (IOException e) {
- hasError = true;
+ recordFailure(e);
log.error("Compactor executor failed due to IOException", e);
throw new HoodieIOException(e.getMessage(), e);
} catch (Exception e) {
- hasError = true;
+ recordFailure(e);
log.error("Compactor executor failed", e);
throw e;
}
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/HoodieAsyncService.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/HoodieAsyncService.java
index 4f207b026e88..aa224a4e4f5d 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/HoodieAsyncService.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/async/HoodieAsyncService.java
@@ -18,6 +18,7 @@
package org.apache.hudi.async;
+import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.Pair;
import lombok.Getter;
@@ -41,8 +42,10 @@ import java.util.function.Function;
public abstract class HoodieAsyncService implements Serializable {
private static final long POLLING_SECONDS = 10;
- // Flag indicating whether an error is incurred in the service
- protected boolean hasError;
+ // Flag indicating whether an error is incurred in the service; volatile as
callers poll it from other threads
+ protected volatile boolean hasError;
+ // Failure the service stopped with, if any
+ private volatile Throwable failure;
// Flag to track if the service is started.
@Getter
private boolean started;
@@ -79,6 +82,21 @@ public abstract class HoodieAsyncService implements
Serializable {
return hasError;
}
+ /**
+ * Returns the failure the service stopped with, if it stopped with one.
+ */
+ public Option<Throwable> getFailure() {
+ return Option.ofNullable(failure);
+ }
+
+ /**
+ * Publish the cause before the error flag so a caller observing the flag
can also read the cause.
+ */
+ protected void recordFailure(Throwable throwable) {
+ failure = throwable;
+ hasError = true;
+ }
+
/**
* Wait till the service shutdown. If the service shutdown with exception,
it will be thrown
*
@@ -134,6 +152,8 @@ public abstract class HoodieAsyncService implements
Serializable {
log.info("The async service already started.");
return;
}
+ failure = null;
+ hasError = false;
Pair<CompletableFuture, ExecutorService> res = startService();
future = res.getKey();
executor = res.getValue();
@@ -157,6 +177,9 @@ public abstract class HoodieAsyncService implements
Serializable {
return;
}
future.whenComplete((resp, error) -> {
+ if (null != error && failure == null) {
+ recordFailure((Throwable) error);
+ }
if (null != callback) {
callback.apply(null != error);
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceFailure.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceFailure.java
new file mode 100644
index 000000000000..3930b34a8e40
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestAsyncTableServiceFailure.java
@@ -0,0 +1,142 @@
+/*
+ * 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.hudi.async;
+
+import org.apache.hudi.client.BaseHoodieWriteClient;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@Timeout(30)
+class TestAsyncTableServiceFailure {
+
+ @ParameterizedTest
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void failureCauseIsPublishedBeforeFutureCompletes(boolean clustering,
boolean ioFailure) throws Exception {
+ HoodieEngineContext context = mock(HoodieEngineContext.class);
+ BaseHoodieWriteClient client = mock(BaseHoodieWriteClient.class);
+ HoodieWriteConfig config = mock(HoodieWriteConfig.class);
+ when(client.getConfig()).thenReturn(config);
+ when(config.areTableServicesEnabled()).thenReturn(true);
+ Exception failure = ioFailure ? new IOException("original failure") : new
IllegalStateException("original failure");
+ CompletionGate gate = new CompletionGate();
+ CountDownLatch callbackFinished = new CountDownLatch(1);
+ HoodieAsyncTableService service;
+ if (clustering) {
+ service = new AsyncClusteringService(context, client) {
+ @Override
+ protected BaseClusterer createClusteringClient(BaseHoodieWriteClient
writeClient) {
+ return new BaseClusterer(writeClient) {
+ @Override
+ public void cluster(String instantTime) throws IOException {
+ throwFailure(failure);
+ }
+ };
+ }
+
+ @Override
+ protected Pair<CompletableFuture, ExecutorService> startService() {
+ return gate.delayCompletion(super.startService());
+ }
+ };
+ } else {
+ service = new AsyncCompactService(context, client) {
+ @Override
+ protected BaseCompactor createCompactor(BaseHoodieWriteClient
writeClient) {
+ return new BaseCompactor(writeClient) {
+ @Override
+ public void compact(String instantTime) throws IOException {
+ throwFailure(failure);
+ }
+ };
+ }
+
+ @Override
+ protected Pair<CompletableFuture, ExecutorService> startService() {
+ return gate.delayCompletion(super.startService());
+ }
+ };
+ }
+ try {
+ service.enqueuePendingAsyncServiceInstant("001");
+ service.start(error -> {
+ callbackFinished.countDown();
+ return true;
+ });
+ await(gate.workerFinished);
+ assertTrue(service.hasError());
+ assertSame(failure, service.getFailure().get());
+ gate.allowCompletion.countDown();
+ await(callbackFinished);
+ assertSame(failure, service.getFailure().get(), "completion must not
replace the original cause with a wrapper");
+ } finally {
+ gate.allowCompletion.countDown();
+ service.shutdown(true);
+ }
+ }
+
+ private static void throwFailure(Exception failure) throws IOException {
+ if (failure instanceof IOException) {
+ throw (IOException) failure;
+ }
+ throw (RuntimeException) failure;
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ assertTrue(latch.await(10, TimeUnit.SECONDS), "timed out waiting for
async service");
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static class CompletionGate {
+ private final CountDownLatch workerFinished = new CountDownLatch(1);
+ private final CountDownLatch allowCompletion = new CountDownLatch(1);
+
+ private Pair<CompletableFuture, ExecutorService>
delayCompletion(Pair<CompletableFuture, ExecutorService> service) {
+ CompletableFuture<?> future = service.getLeft();
+ return Pair.of(future.handleAsync((result, error) -> {
+ workerFinished.countDown();
+ await(allowCompletion);
+ if (error != null) {
+ throw new CompletionException(error);
+ }
+ return result;
+ }, service.getRight()), service.getRight());
+ }
+ }
+}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestHoodieAsyncTableService.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestHoodieAsyncTableService.java
index 0c19576d042b..ef2ca1a88d49 100644
---
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestHoodieAsyncTableService.java
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/async/TestHoodieAsyncTableService.java
@@ -24,18 +24,42 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
class TestHoodieAsyncTableService {
+ @Test
+ void futureFailureIsRecordedWithoutUnwrapping() {
+ CompletionException failure = new CompletionException(new
IllegalStateException("original failure"));
+ CompletableFuture<Boolean> future = new CompletableFuture<>();
+ future.completeExceptionally(failure);
+ HoodieAsyncService service = new HoodieAsyncService() {
+ @Override
+ protected Pair<CompletableFuture, ExecutorService> startService() {
+ return Pair.of(future, null);
+ }
+ };
+
+ service.start(null);
+
+ assertTrue(service.hasError());
+ assertSame(failure, service.getFailure().get());
+ }
+
@Test
void tableServiceShouldNotStartIfDisabled(@Mock HoodieWriteConfig config) {
when(config.areTableServicesEnabled()).thenReturn(false);
@@ -44,6 +68,38 @@ class TestHoodieAsyncTableService {
assertFalse(service.isStarted());
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void restartingServiceClearsPreviousFailure(boolean failAgain) {
+ AtomicReference<CompletableFuture<Boolean>> next = new
AtomicReference<>(new CompletableFuture<>());
+ HoodieAsyncService service = new HoodieAsyncService() {
+ @Override
+ protected Pair<CompletableFuture, ExecutorService> startService() {
+ return Pair.of(next.get(), null);
+ }
+ };
+ RuntimeException firstFailure = new IllegalStateException("first run
failed");
+ next.get().completeExceptionally(firstFailure);
+ service.start(null);
+ assertSame(firstFailure, service.getFailure().get());
+ assertFalse(service.isStarted());
+
+ next.set(new CompletableFuture<>());
+ service.start(null);
+ assertFalse(service.hasError());
+ assertFalse(service.getFailure().isPresent());
+ if (failAgain) {
+ RuntimeException secondFailure = new IllegalStateException("second run
failed");
+ next.get().completeExceptionally(secondFailure);
+ assertTrue(service.hasError());
+ assertSame(secondFailure, service.getFailure().get());
+ } else {
+ next.get().complete(true);
+ assertFalse(service.hasError());
+ assertFalse(service.getFailure().isPresent());
+ }
+ }
+
private static class DummyAsyncTableService extends HoodieAsyncTableService {
protected DummyAsyncTableService(HoodieWriteConfig writeConfig) {
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
index d062e11b6bc2..8bc744876168 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamer.java
@@ -842,6 +842,7 @@ public class HoodieStreamer implements Serializable {
while (!isShutdownRequested()) {
try {
long start = System.currentTimeMillis();
+ throwIfAsyncTableServiceFailed();
// Send a heartbeat metrics event to track the active ingestion
job for this table.
streamSync.getMetrics().updateStreamerHeartbeatTimestamp(start);
// check if deltastreamer need to update the configuration
before the sync
@@ -860,10 +861,7 @@ public class HoodieStreamer implements Serializable {
log.info("Enqueuing new pending compaction instant ({})",
scheduledCompactionInstantAndRDD.get().getLeft());
asyncCompactService.get().enqueuePendingAsyncServiceInstant(scheduledCompactionInstantAndRDD.get().getLeft().get());
asyncCompactService.get().waitTillPendingAsyncServiceInstantsReducesTo(cfg.maxPendingCompactions);
- if (asyncCompactService.get().hasError()) {
- error = true;
- throw new HoodieException("Async compaction failed.
Shutting down Delta Sync...");
- }
+ throwIfAsyncTableServiceFailed();
}
if (clusteringConfig.isAsyncClusteringEnabled()) {
Option<String> clusteringInstant =
streamSync.getClusteringInstantOpt();
@@ -871,10 +869,7 @@ public class HoodieStreamer implements Serializable {
log.info("Scheduled async clustering for instant: {}",
clusteringInstant.get());
asyncClusteringService.get().enqueuePendingAsyncServiceInstant(clusteringInstant.get());
asyncClusteringService.get().waitTillPendingAsyncServiceInstantsReducesTo(cfg.maxPendingClustering);
- if (asyncClusteringService.get().hasError()) {
- error = true;
- throw new HoodieException("Async clustering failed.
Shutting down Delta Sync...");
- }
+ throwIfAsyncTableServiceFailed();
}
}
// check if deltastreamer need to be shutdown
@@ -904,6 +899,22 @@ public class HoodieStreamer implements Serializable {
}, executor), executor);
}
+ /**
+ * Stops ingestion when an async table service stopped with an error, with
that error as the cause.
+ * Checked on every round of ingestion, not only when a new compaction or
+ * clustering is scheduled, so that ingestion does not go on without its
table services.
+ */
+ private void throwIfAsyncTableServiceFailed() {
+ if (asyncCompactService.isPresent() &&
asyncCompactService.get().hasError()) {
+ throw new HoodieException("Async compaction failed. Shutting down
Delta Sync...",
+ asyncCompactService.get().getFailure().orElse(null));
+ }
+ if (asyncClusteringService.isPresent() &&
asyncClusteringService.get().hasError()) {
+ throw new HoodieException("Async clustering failed. Shutting down
Delta Sync...",
+ asyncClusteringService.get().getFailure().orElse(null));
+ }
+ }
+
private void handleUpsertException(HoodieUpsertException ue) {
if (ue.getCause() instanceof HoodieClusteringUpdateException) {
log.warn("Write rejected due to conflicts with pending clustering
operation. Going to retry after 1 min with the hope "
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
index 1ec5298436ff..262a58bd8908 100644
---
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
+++
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/StreamSync.java
@@ -1247,13 +1247,19 @@ public class StreamSync implements Serializable,
Closeable {
}
}
- if (writeClient != null) {
- // Close Write client.
- writeClient.close();
- }
KeyGenUtils.recordComplexKeygenEncodingIfMissing(metaClient, writeConfig);
+ SparkRDDWriteClient previousWriteClient = writeClient;
writeClient = new SparkRDDWriteClient<>(hoodieSparkContext, writeConfig,
embeddedTimelineService);
- onInitializingHoodieWriteClient.apply(writeClient);
+ // Switch the async table services before closing the previous client, so
new runs use the new client. The switch
+ // does not wait for an active run; a run still using the previous client
may fail once it is closed, and that
+ // failure stops ingestion with its cause.
+ try {
+ onInitializingHoodieWriteClient.apply(writeClient);
+ } finally {
+ if (previousWriteClient != null) {
+ previousWriteClient.close();
+ }
+ }
}
protected void releaseResources(String instantTime) {
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieStreamerAsyncTableServiceFailure.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieStreamerAsyncTableServiceFailure.java
new file mode 100644
index 000000000000..66e44f2cd1f9
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/deltastreamer/TestHoodieStreamerAsyncTableServiceFailure.java
@@ -0,0 +1,131 @@
+/*
+ * 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.hudi.utilities.deltastreamer;
+
+import org.apache.hudi.async.AsyncCompactService;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.model.OverwriteWithLatestAvroPayload;
+import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.utilities.config.SourceTestConfig;
+import org.apache.hudi.utilities.sources.TestDataSource;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.spark.TaskContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Failures of the async table services of a continuous {@link
HoodieDeltaStreamer}.
+ */
+public class TestHoodieStreamerAsyncTableServiceFailure extends
HoodieDeltaStreamerTestBase {
+
+ @AfterEach
+ public void resetFailureInjection() {
+ CompactionFailingPayload.failCompaction = false;
+ TestDataSource.returnEmptyBatch = false;
+ }
+
+ /**
+ * When async compaction fails with an error its service cannot recover
from, the streamer stops with that error as
+ * the cause. It used to notice only when it next scheduled a compaction,
and then without the cause.
+ */
+ @Test
+ public void testStreamerStopsWithCauseWhenAsyncCompactionFails() throws
Exception {
+ String tableBasePath = basePath + "/async_compaction_failure";
+ HoodieDeltaStreamer.Config cfg = TestHelpers.makeConfig(tableBasePath,
WriteOperationType.UPSERT,
+
Collections.singletonList(TestHoodieDeltaStreamer.TripsWithDistanceTransformer.class.getName()),
+ PROPS_FILENAME_TEST_SOURCE, false, true, true,
CompactionFailingPayload.class.getName(),
+ HoodieTableType.MERGE_ON_READ.name());
+ cfg.continuousMode = true;
+ cfg.configs.add(String.format("%s=%d",
SourceTestConfig.MAX_UNIQUE_RECORDS_PROP.key(), 1000));
+ cfg.configs.add(String.format("%s=%d",
HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(), 2));
+ CompactionFailingPayload.failCompaction = true;
+
+ HoodieDeltaStreamer streamer = new HoodieDeltaStreamer(cfg, jsc);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> sync = executor.submit(() -> {
+ streamer.sync();
+ return null;
+ });
+ ExecutionException failure = assertThrows(ExecutionException.class, ()
-> sync.get(5, TimeUnit.MINUTES),
+ "the streamer must stop when async compaction fails");
+ assertTrue(causeChainContains(failure, "Async compaction failed"),
"unexpected failure: " + failure);
+ assertTrue(causeChainContains(failure,
CompactionFailingPayload.FAILURE_MESSAGE),
+ "the compaction failure must be the cause: " + failure);
+ } finally {
+ streamer.shutdownGracefully();
+ executor.shutdownNow();
+ }
+ }
+
+ private static boolean causeChainContains(Throwable throwable, String
message) {
+ for (Throwable t = throwable; t != null; t = t.getCause()) {
+ if (t.getMessage() != null && t.getMessage().contains(message)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Payload whose merges fail inside async compaction, i.e. in Spark tasks of
the compaction scheduler pool, while
+ * {@link #failCompaction} is set.
+ */
+ public static class CompactionFailingPayload extends
OverwriteWithLatestAvroPayload {
+
+ static final String FAILURE_MESSAGE = "Injected async compaction failure";
+ static volatile boolean failCompaction = false;
+
+ public CompactionFailingPayload(GenericRecord record, Comparable
orderingVal) {
+ super(record, orderingVal);
+ }
+
+ public CompactionFailingPayload(Option<GenericRecord> record) {
+ super(record);
+ }
+
+ @Override
+ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord
currentValue, Schema schema) throws IOException {
+ TaskContext taskContext = TaskContext.get();
+ if (failCompaction && taskContext != null
+ &&
AsyncCompactService.COMPACT_POOL_NAME.equals(taskContext.getLocalProperty("spark.scheduler.pool")))
{
+ throw new HoodieException(FAILURE_MESSAGE);
+ }
+ return super.combineAndGetUpdateValue(currentValue, schema);
+ }
+ }
+}
