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 bc78d06bd217 fix(kafka-connect): treat a completed commit as
successful when a post-commit action fails (#20316)
bc78d06bd217 is described below
commit bc78d06bd2179b4304208b6aae23ce0fe6c9ad72
Author: Vova Kolmakov <[email protected]>
AuthorDate: Thu Oct 15 10:03:53 2026 +0700
fix(kafka-connect): treat a completed commit as successful when a
post-commit action fails (#20316)
Co-authored-by: Vova Kolmakov <[email protected]>
---
.../writers/KafkaConnectTransactionServices.java | 51 +++++++---
.../TestKafkaConnectTransactionServices.java | 108 ++++++++++++++++++++-
2 files changed, 143 insertions(+), 16 deletions(-)
diff --git
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
index 6b89627fad9c..77ba267129ee 100644
---
a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
+++
b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectTransactionServices.java
@@ -119,24 +119,43 @@ public class KafkaConnectTransactionServices implements
ConnectTransactionServic
@Override
public boolean endCommit(String commitTime, List<WriteStatus> writeStatuses,
Map<String, String> extraMetadata) {
- boolean success = javaClient.commit(commitTime, writeStatuses,
Option.of(extraMetadata));
+ boolean success;
+ try {
+ success = javaClient.commit(commitTime, writeStatuses,
Option.of(extraMetadata));
+ } catch (RuntimeException exception) {
+ // Once the instant is completed, its data and Kafka offsets are
visible, so a failure after that point inside the
+ // commit (e.g. in cleaning or archiving) must not fail it, or the
participants would write the same records again
+ if (!isCommitCompleted(commitTime)) {
+ throw exception;
+ }
+ log.error("Commit {} completed, but a post-commit action failed",
commitTime, exception);
+ success = true;
+ }
if (success) {
log.info("Ending Hudi commit {}", commitTime);
-
- // Schedule clustering and compaction as needed.
- if (writeConfig.isAsyncClusteringEnabled()) {
- javaClient.scheduleClustering(Option.empty()).ifPresent(
- instantTs -> log.info("Scheduled clustering at instant time:{}",
instantTs));
+ try {
+ scheduleTableServicesAndSyncMeta();
+ } catch (RuntimeException exception) {
+ // Likewise, these follow-up actions must not fail the commit; they
run again after the next one
+ log.error("Commit {} succeeded, but scheduling table services or meta
sync failed", commitTime, exception);
}
- if (isAsyncCompactionEnabled()) {
- javaClient.scheduleCompaction(Option.empty()).ifPresent(
- instantTs -> log.info("Scheduled compaction at instant time:{}",
instantTs));
- }
- syncMeta();
}
return success;
}
+ private void scheduleTableServicesAndSyncMeta() {
+ // Schedule clustering and compaction as needed.
+ if (writeConfig.isAsyncClusteringEnabled()) {
+ javaClient.scheduleClustering(Option.empty()).ifPresent(
+ instantTs -> log.info("Scheduled clustering at instant time:{}",
instantTs));
+ }
+ if (isAsyncCompactionEnabled()) {
+ javaClient.scheduleCompaction(Option.empty()).ifPresent(
+ instantTs -> log.info("Scheduled compaction at instant time:{}",
instantTs));
+ }
+ syncMeta();
+ }
+
@Override
public Map<String, String> fetchLatestExtraCommitMetadata() {
if (tableMetaClient.isPresent()) {
@@ -162,6 +181,16 @@ public class KafkaConnectTransactionServices implements
ConnectTransactionServic
&& connectConfigs.isAsyncCompactEnabled();
}
+ private boolean isCommitCompleted(String commitTime) {
+ try {
+ return tableMetaClient.isPresent()
+ &&
tableMetaClient.get().reloadActiveTimeline().filterCompletedInstants().containsInstant(commitTime);
+ } catch (RuntimeException exception) {
+ log.warn("Failed to check whether commit {} completed", commitTime,
exception);
+ return false;
+ }
+ }
+
private void syncMeta() {
if (connectConfigs.isMetaSyncEnabled()) {
Set<String> syncClientToolClasses = new HashSet<>(
diff --git
a/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestKafkaConnectTransactionServices.java
b/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestKafkaConnectTransactionServices.java
index b561104c6a96..7b450338be68 100644
---
a/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestKafkaConnectTransactionServices.java
+++
b/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestKafkaConnectTransactionServices.java
@@ -23,18 +23,29 @@ package org.apache.hudi.writers;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.timeline.HoodieTimeline;
+import org.apache.hudi.common.testutils.FileCreateUtils;
import org.apache.hudi.common.testutils.HoodieTestUtils;
+import org.apache.hudi.config.HoodieClusteringConfig;
import org.apache.hudi.connect.writers.KafkaConnectConfigs;
import org.apache.hudi.connect.writers.KafkaConnectTransactionServices;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import org.apache.hudi.sync.common.HoodieSyncTool;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import java.io.IOException;
import java.nio.file.Path;
+import java.util.Collections;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class TestKafkaConnectTransactionServices {
@@ -43,11 +54,7 @@ public class TestKafkaConnectTransactionServices {
@Test
public void testTableCreation() {
- Properties props = new Properties();
- props.put("hoodie.table.name", "test");
- props.put("hoodie.base.path", path);
- props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(),
"partition_path");
- props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "key");
+ Properties props = tableProps();
props.put("hoodie.table.version", "6");
KafkaConnectConfigs configs = KafkaConnectConfigs.newBuilder()
.withProperties(props)
@@ -58,4 +65,95 @@ public class TestKafkaConnectTransactionServices {
assertEquals(HoodieTableVersion.SIX,
metaClient.getTableConfig().getTableVersion());
assertEquals(HoodieTableType.COPY_ON_WRITE,
metaClient.getTableConfig().getTableType());
}
+
+ @Test
+ public void testEndCommitSucceedsWhenMetaSyncFailsAfterCompletion() {
+ Properties props = tableProps();
+ props.put(KafkaConnectConfigs.META_SYNC_ENABLE.key(), "true");
+ props.put(KafkaConnectConfigs.META_SYNC_CLASSES.key(),
FailingMetaSyncTool.class.getName());
+ FailingMetaSyncTool.SYNC_ATTEMPTS.set(0);
+ try (KafkaConnectTransactionServices transactionServices = new
KafkaConnectTransactionServices(
+ KafkaConnectConfigs.newBuilder().withProperties(props).build())) {
+ // Like ConnectTransactionCoordinator.start(), cache the timeline before
the commit, so endCommit has to reload it
+ transactionServices.fetchLatestExtraCommitMetadata();
+ String commitTime = transactionServices.startCommit();
+
+ // The instant completes and only the meta sync after it fails, so the
commit must succeed
+ assertTrue(transactionServices.endCommit(commitTime,
Collections.emptyList(),
+ Collections.singletonMap("kafka.commit.offsets", "0=10")));
+ assertEquals(1, FailingMetaSyncTool.SYNC_ATTEMPTS.get());
+ assertTrue(isCompleted(commitTime));
+ }
+ }
+
+ @Test
+ public void testEndCommitSucceedsWhenInlineClusteringFailsAfterCompletion() {
+ Properties props = tableProps();
+ props.put(HoodieClusteringConfig.INLINE_CLUSTERING.key(), "true");
+ props.put(HoodieClusteringConfig.INLINE_CLUSTERING_MAX_COMMITS.key(), "1");
+ props.put(HoodieClusteringConfig.PLAN_STRATEGY_CLASS_NAME.key(),
"org.apache.hudi.NonExistentClusteringPlanStrategy");
+ props.put(KafkaConnectConfigs.META_SYNC_ENABLE.key(), "true");
+ props.put(KafkaConnectConfigs.META_SYNC_CLASSES.key(),
FailingMetaSyncTool.class.getName());
+ FailingMetaSyncTool.SYNC_ATTEMPTS.set(0);
+ try (KafkaConnectTransactionServices transactionServices = new
KafkaConnectTransactionServices(
+ KafkaConnectConfigs.newBuilder().withProperties(props).build())) {
+ transactionServices.fetchLatestExtraCommitMetadata();
+ String commitTime = transactionServices.startCommit();
+
+ // Like cleaning and archiving, inline clustering runs inside
javaClient.commit after the instant completed;
+ // it fails to load its plan strategy, so the commit must still succeed
and the meta sync must still run
+ assertTrue(transactionServices.endCommit(commitTime,
Collections.emptyList(),
+ Collections.singletonMap("kafka.commit.offsets", "0=10")));
+ assertEquals(1, FailingMetaSyncTool.SYNC_ATTEMPTS.get());
+ assertTrue(isCompleted(commitTime));
+ }
+ }
+
+ @Test
+ public void testEndCommitFailsWhenInstantDoesNotComplete() throws
IOException {
+ try (KafkaConnectTransactionServices transactionServices = new
KafkaConnectTransactionServices(
+
KafkaConnectConfigs.newBuilder().withProperties(tableProps()).build())) {
+ String commitTime = transactionServices.startCommit();
+ // Losing the inflight file makes the commit fail while its instant is
still pending
+
FileCreateUtils.deleteInflightCommit(HoodieTestUtils.createMetaClient(path.toFile().getPath()),
commitTime);
+
+ assertThrows(RuntimeException.class, () ->
transactionServices.endCommit(commitTime, Collections.emptyList(),
+ Collections.emptyMap()));
+ HoodieTimeline timeline =
HoodieTestUtils.createMetaClient(path.toFile().getPath()).getActiveTimeline();
+
assertTrue(timeline.filterInflightsAndRequested().containsInstant(commitTime));
+
assertFalse(timeline.filterCompletedInstants().containsInstant(commitTime));
+ }
+ }
+
+ private boolean isCompleted(String commitTime) {
+ return
HoodieTestUtils.createMetaClient(path.toFile().getPath()).getActiveTimeline()
+ .filterCompletedInstants().containsInstant(commitTime);
+ }
+
+ private Properties tableProps() {
+ Properties props = new Properties();
+ props.put("hoodie.table.name", "test");
+ props.put("hoodie.base.path", path);
+ props.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(),
"partition_path");
+ props.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "key");
+ return props;
+ }
+
+ /**
+ * A meta sync tool that fails like an unreachable metastore.
+ */
+ public static class FailingMetaSyncTool extends HoodieSyncTool {
+
+ static final AtomicInteger SYNC_ATTEMPTS = new AtomicInteger();
+
+ public FailingMetaSyncTool(Properties props) {
+ super(props);
+ }
+
+ @Override
+ public void syncHoodieTable() {
+ SYNC_ATTEMPTS.incrementAndGet();
+ throw new HoodieException("Metastore unavailable");
+ }
+ }
}
