This is an automated email from the ASF dual-hosted git repository.
voonhous 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 79fedcf12f43 fix(client): report completed timeline action in
clustering callback (#19464)
79fedcf12f43 is described below
commit 79fedcf12f43ca7504c7a40df5ef2c9f8cfd7cf8
Author: Sagar Sumit <[email protected]>
AuthorDate: Thu Aug 6 18:39:18 2026 +0530
fix(client): report completed timeline action in clustering callback
(#19464)
* fix(client): report the completed timeline action in the clustering write
commit callback
Signed-off-by: codope <[email protected]>
* test(client): pin completed clustering action, share callback
Signed-off-by: codope <[email protected]>
* fix(client): use REPLACE_COMMIT_ACTION constant in clustering callback
Signed-off-by: codope <[email protected]>
---------
Signed-off-by: codope <[email protected]>
---
.../hudi/client/BaseHoodieTableServiceClient.java | 2 +-
.../hudi/testutils/RecordingCommitCallback.java | 56 ++++++++++++++++++++++
.../hudi/client/HoodieFlinkTableServiceClient.java | 2 +-
.../client/TestHoodieFlinkTableServiceClient.java | 14 ++++++
.../TestHoodieJavaClientOnMergeOnReadStorage.java | 41 ++++------------
.../apache/hudi/common/util/ClusteringUtils.java | 6 +--
.../hudi/common/util/TestClusteringUtils.java | 43 +++++++++++++++++
7 files changed, 128 insertions(+), 36 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
index 235b1023cdae..06020e059199 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieTableServiceClient.java
@@ -644,7 +644,7 @@ public abstract class BaseHoodieTableServiceClient<I, T, O>
extends BaseHoodieCl
heartbeatClient.stop(clusteringCommitTime);
}
log.info("Clustering successfully on commit {} for table {}",
clusteringCommitTime, table.getConfig().getBasePath());
- fireCommitCallbackIfNecessary(clusteringCommitTime,
clusteringInstant.getAction(),
+ fireCommitCallbackIfNecessary(clusteringCommitTime,
HoodieTimeline.REPLACE_COMMIT_ACTION,
writeStats, table::getBaseFileOnlyView, Option.empty());
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/RecordingCommitCallback.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/RecordingCommitCallback.java
new file mode 100644
index 000000000000..0d5944fcfdc5
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/RecordingCommitCallback.java
@@ -0,0 +1,56 @@
+/*
+ * 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.testutils;
+
+import org.apache.hudi.callback.HoodieWriteCommitCallback;
+import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * A recording {@link HoodieWriteCommitCallback} that captures every fired
message so tests can
+ * assert the callback fires for table-service (compaction/clustering) commits
with the expected
+ * action type. Loaded reflectively from the write config, so it needs a public
+ * {@code (HoodieWriteConfig)} constructor, and the messages have to live in
static state; call
+ * {@link #reset()} at the start of every test that asserts on them.
+ */
+public class RecordingCommitCallback implements HoodieWriteCommitCallback {
+
+ private static final List<HoodieWriteCommitCallbackMessage> MESSAGES = new
CopyOnWriteArrayList<>();
+
+ public RecordingCommitCallback(HoodieWriteConfig config) {
+ // config arg required for reflective instantiation
+ }
+
+ @Override
+ public void call(HoodieWriteCommitCallbackMessage callbackMessage) {
+ MESSAGES.add(callbackMessage);
+ }
+
+ public static List<HoodieWriteCommitCallbackMessage> messages() {
+ return new ArrayList<>(MESSAGES);
+ }
+
+ public static void reset() {
+ MESSAGES.clear();
+ }
+}
diff --git
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
index 0ab62d498ddd..579861677a14 100644
---
a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
+++
b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkTableServiceClient.java
@@ -162,7 +162,7 @@ public class HoodieFlinkTableServiceClient<T> extends
BaseHoodieTableServiceClie
}
}
log.info("Clustering successfully on commit {}", clusteringCommitTime);
- fireCommitCallbackIfNecessary(clusteringCommitTime,
clusteringInstant.getAction(),
+ fireCommitCallbackIfNecessary(clusteringCommitTime,
HoodieActiveTimeline.REPLACE_COMMIT_ACTION,
writeStats, table::getBaseFileOnlyView, Option.empty());
}
diff --git
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java
index 25d12795e465..7ddacba98a66 100644
---
a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java
+++
b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestHoodieFlinkTableServiceClient.java
@@ -18,6 +18,7 @@
package org.apache.hudi.client;
+import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
import org.apache.hudi.client.common.HoodieFlinkEngineContext;
import org.apache.hudi.client.embedded.EmbeddedTimelineService;
import org.apache.hudi.common.config.HoodieMetadataConfig;
@@ -31,6 +32,7 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.ClusteringUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieLockConfig;
+import org.apache.hudi.config.HoodieWriteCommitCallbackConfig;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.core.transaction.lock.InProcessLockProvider;
import org.apache.hudi.exception.HoodieException;
@@ -43,6 +45,7 @@ import org.apache.hudi.table.action.compact.CompactHelpers;
import org.apache.hudi.table.marker.WriteMarkers;
import org.apache.hudi.table.marker.WriteMarkersFactory;
import org.apache.hudi.testutils.HoodieFlinkClientTestHarness;
+import org.apache.hudi.testutils.RecordingCommitCallback;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -53,7 +56,9 @@ import org.mockito.Mockito;
import java.io.IOException;
import java.util.Collections;
+import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
@@ -218,15 +223,21 @@ public class TestHoodieFlinkTableServiceClient extends
HoodieFlinkClientTestHarn
@Test
void testCompleteClusteringCommitsAndCleansMarkers() {
+ RecordingCommitCallback.reset();
HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder()
.withPath(metaClient.getBasePath())
.withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build())
+ .withCallbackConfig(HoodieWriteCommitCallbackConfig.newBuilder()
+ .writeCommitCallbackOn("true")
+ .withCallbackClass(RecordingCommitCallback.class.getName())
+ .build())
.build();
HoodieFlinkTable table = mock(HoodieFlinkTable.class);
HoodieActiveTimeline activeTimeline = mock(HoodieActiveTimeline.class);
when(table.getActiveTimeline()).thenReturn(activeTimeline);
when(table.getInstantGenerator()).thenReturn(metaClient.getInstantGenerator());
HoodieInstant clusteringInstant = mock(HoodieInstant.class);
+
when(clusteringInstant.getAction()).thenReturn(HoodieTimeline.CLUSTERING_ACTION);
HoodieReplaceCommitMetadata metadata = new HoodieReplaceCommitMetadata();
TestableHoodieFlinkTableServiceClient client =
new TestableHoodieFlinkTableServiceClient(context, writeConfig,
Option.empty(), table);
@@ -244,6 +255,9 @@ public class TestHoodieFlinkTableServiceClient extends
HoodieFlinkClientTestHarn
}
verify(writeMarkers).quietDeleteMarkerDir(any(), any(Integer.class));
+ List<HoodieWriteCommitCallbackMessage> messages =
RecordingCommitCallback.messages();
+ assertEquals(1, messages.size(), "callback must fire once for the
clustering commit");
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION,
messages.get(0).getCommitActionType().orElse(null));
}
private static class TestableHoodieFlinkTableServiceClient extends
HoodieFlinkTableServiceClient<Object> {
diff --git
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
index 3964dd69a517..74185cc35cea 100644
---
a/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
+++
b/hudi-client/hudi-java-client/src/test/java/org/apache/hudi/client/functional/TestHoodieJavaClientOnMergeOnReadStorage.java
@@ -18,7 +18,6 @@
package org.apache.hudi.client.functional;
-import org.apache.hudi.callback.HoodieWriteCommitCallback;
import org.apache.hudi.callback.common.HoodieWriteCommitCallbackMessage;
import org.apache.hudi.client.HoodieJavaWriteClient;
import org.apache.hudi.client.WriteClientTestUtils;
@@ -38,6 +37,7 @@ import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.table.action.HoodieWriteMetadata;
import org.apache.hudi.testutils.GenericRecordValidationTestUtils;
import org.apache.hudi.testutils.HoodieJavaClientTestHarness;
+import org.apache.hudi.testutils.RecordingCommitCallback;
import org.apache.avro.generic.GenericRecord;
import org.junit.jupiter.api.BeforeEach;
@@ -46,7 +46,6 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;
import static
org.apache.hudi.common.testutils.HoodieTestUtils.INSTANT_GENERATOR;
@@ -193,7 +192,7 @@ public class TestHoodieJavaClientOnMergeOnReadStorage
extends HoodieJavaClientTe
@Test
public void testWriteCommitCallbackFiresOnCompaction() throws Exception {
- RecordingCommitCallback.MESSAGES.clear();
+ RecordingCommitCallback.reset();
HoodieWriteConfig config =
getConfigBuilder(HoodieTestDataGenerator.TRIP_EXAMPLE_SCHEMA,
HoodieIndex.IndexType.INMEMORY)
.withCompactionConfig(HoodieCompactionConfig.newBuilder().withMaxNumDeltaCommitsBeforeCompaction(2).build())
@@ -215,7 +214,7 @@ public class TestHoodieJavaClientOnMergeOnReadStorage
extends HoodieJavaClientTe
false, false, 5, 100, 2, config.populateMetaFields(),
INSTANT_GENERATOR);
// The callback must fire for the auto-committed delta commits with the
deltacommit action.
- assertTrue(RecordingCommitCallback.MESSAGES.stream().anyMatch(m ->
+ assertTrue(RecordingCommitCallback.messages().stream().anyMatch(m ->
HoodieTimeline.DELTA_COMMIT_ACTION.equals(m.getCommitActionType().orElse(null))),
"callback must fire for delta commits");
@@ -228,7 +227,7 @@ public class TestHoodieJavaClientOnMergeOnReadStorage
extends HoodieJavaClientTe
// The callback must fire exactly once for the compaction completion,
reporting the completed
// timeline action (commit).
- List<HoodieWriteCommitCallbackMessage> compactionMessages =
RecordingCommitCallback.MESSAGES.stream()
+ List<HoodieWriteCommitCallbackMessage> compactionMessages =
RecordingCommitCallback.messages().stream()
.filter(m -> m.getCommitTime().equals(compactionTime.get()))
.collect(Collectors.toList());
assertEquals(1, compactionMessages.size(), "callback must fire once for
the compaction commit");
@@ -238,7 +237,7 @@ public class TestHoodieJavaClientOnMergeOnReadStorage
extends HoodieJavaClientTe
@Test
public void testWriteCommitCallbackFiresOnClustering() throws Exception {
- RecordingCommitCallback.MESSAGES.clear();
+ RecordingCommitCallback.reset();
HoodieClusteringConfig clusteringConfig =
HoodieClusteringConfig.newBuilder()
.withClusteringMaxNumGroups(10)
.withClusteringSortColumns("_row_key")
@@ -271,34 +270,14 @@ public class TestHoodieJavaClientOnMergeOnReadStorage
extends HoodieJavaClientTe
assertTrue(metaClient.reloadActiveTimeline().filterCompletedInstants().containsInstant(clusteringTime.get()));
// The callback must fire once for the clustering completion, reporting
the action actually on
- // the timeline (replacecommit for table version < 8, clustering for 8+).
- List<HoodieWriteCommitCallbackMessage> clusteringMessages =
RecordingCommitCallback.MESSAGES.stream()
+ // the timeline (replacecommit).
+ List<HoodieWriteCommitCallbackMessage> clusteringMessages =
RecordingCommitCallback.messages().stream()
.filter(m -> m.getCommitTime().equals(clusteringTime.get()))
.collect(Collectors.toList());
assertEquals(1, clusteringMessages.size(), "callback must fire once for
the clustering commit");
- String action =
clusteringMessages.get(0).getCommitActionType().orElse(null);
- assertTrue(HoodieTimeline.REPLACE_COMMIT_ACTION.equals(action) ||
HoodieTimeline.CLUSTERING_ACTION.equals(action),
- "clustering callback must report the timeline action, got: " + action);
- }
-
- /**
- * A recording {@link HoodieWriteCommitCallback} that captures every fired
message so tests can
- * assert the callback fires for table-service (compaction/clustering)
commits with the expected
- * action type. Loaded reflectively from the write config, so it needs a
public
- * {@code (HoodieWriteConfig)} constructor.
- */
- public static class RecordingCommitCallback implements
HoodieWriteCommitCallback {
-
- static final List<HoodieWriteCommitCallbackMessage> MESSAGES = new
CopyOnWriteArrayList<>();
-
- public RecordingCommitCallback(HoodieWriteConfig config) {
- // config arg required for reflective instantiation
- }
-
- @Override
- public void call(HoodieWriteCommitCallbackMessage callbackMessage) {
- MESSAGES.add(callbackMessage);
- }
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION,
clusteringMessages.get(0).getCommitActionType().orElse(null));
+ // Clustering writes new file groups, so every stat carries NULL_COMMIT
and no prev path resolves.
+ assertTrue(clusteringMessages.get(0).getPrevFilePaths().isEmpty());
}
}
diff --git
a/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
b/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
index a7c4609ffa37..f623fbcfaf50 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ClusteringUtils.java
@@ -122,9 +122,9 @@ public class ClusteringUtils {
* Transitions the provided clustering instant fron inflight to complete
based on the clustering
* action type. After HUDI-7905, the new clustering commits are written with
clustering action.
*/
- public static <T> void
transitionClusteringOrReplaceInflightToComplete(boolean shouldLock,
HoodieInstant clusteringInstant,
-
HoodieReplaceCommitMetadata metadata, HoodieActiveTimeline activeTimeline,
-
TableFormatCompletionAction tableFormatCompletionAction) {
+ public static void transitionClusteringOrReplaceInflightToComplete(boolean
shouldLock, HoodieInstant clusteringInstant,
+
HoodieReplaceCommitMetadata metadata, HoodieActiveTimeline activeTimeline,
+
TableFormatCompletionAction tableFormatCompletionAction) {
if
(clusteringInstant.getAction().equals(HoodieTimeline.CLUSTERING_ACTION)) {
activeTimeline.transitionClusterInflightToComplete(shouldLock,
clusteringInstant, metadata, tableFormatCompletionAction);
} else {
diff --git
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
index 6d580e3b5c56..25490d3971dd 100644
---
a/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
+++
b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
@@ -124,10 +124,53 @@ public class TestClusteringUtils extends
HoodieCommonTestHarness {
//now that it is complete, the first instant should be picked
HoodieInstant complete =
metaClient.getActiveTimeline().transitionClusterInflightToComplete(false,
inflight, new HoodieReplaceCommitMetadata());
assertEquals(HoodieInstant.State.COMPLETED, complete.getState());
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION, complete.getAction());
lastPendingClustering =
metaClient.reloadActiveTimeline().getLastPendingClusterInstant();
assertEquals("1", lastPendingClustering.get().requestedTime());
}
+ /**
+ * A clustering commit completes as a {@code replacecommit} whichever action
its inflight instant
+ * carried, {@code clustering} on table version 8+ or {@code replacecommit}
before that.
+ */
+ @Test
+ public void testClusteringAndReplaceInflightBothCompleteAsReplaceCommit()
throws Exception {
+ List<String> fileIds = new ArrayList<>();
+ fileIds.add(UUID.randomUUID().toString());
+ List<HoodieInstant> completed = new ArrayList<>();
+
+ HoodieInstant clusterRequested =
createRequestedClusterInstant("partition1", "1", fileIds);
+ HoodieInstant clusterInflight = metaClient.getActiveTimeline()
+ .transitionClusterRequestedToInflight(clusterRequested,
Option.empty());
+ assertEquals(HoodieTimeline.CLUSTERING_ACTION,
clusterInflight.getAction());
+ ClusteringUtils.transitionClusteringOrReplaceInflightToComplete(
+ false, clusterInflight, new HoodieReplaceCommitMetadata(),
metaClient.getActiveTimeline(),
+ completed::add);
+ assertEquals(1, completed.size());
+ assertEquals(HoodieInstant.State.COMPLETED, completed.get(0).getState());
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION,
completed.get(0).getAction(),
+ "a clustering inflight instant must complete as replacecommit");
+
+ completed.clear();
+ HoodieInstant replaceRequested =
createRequestedReplaceInstantNotClustering("2");
+ HoodieInstant replaceInflight = metaClient.getActiveTimeline()
+ .transitionReplaceRequestedToInflight(replaceRequested,
Option.empty());
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION,
replaceInflight.getAction());
+ ClusteringUtils.transitionClusteringOrReplaceInflightToComplete(
+ false, replaceInflight, new HoodieReplaceCommitMetadata(),
metaClient.getActiveTimeline(),
+ completed::add);
+ assertEquals(1, completed.size());
+ assertEquals(HoodieInstant.State.COMPLETED, completed.get(0).getState());
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION,
completed.get(0).getAction(),
+ "a replacecommit inflight instant must complete as replacecommit");
+
+ List<HoodieInstant> completedInstants =
+
metaClient.reloadActiveTimeline().filterCompletedInstants().getInstants();
+ assertEquals(2, completedInstants.size());
+ completedInstants.forEach(instant ->
+ assertEquals(HoodieTimeline.REPLACE_COMMIT_ACTION,
instant.getAction()));
+ }
+
// replacecommit.inflight doesn't have clustering plan.
// Verify that getClusteringPlan fetches content from corresponding
requested file.
@Disabled("Will fail due to avro issue AVRO-3789. This is fixed in avro
1.11.3")
