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 697bfe1d000d fix(flink): sort ClientIds heartbeat files numerically
instead of lexicographically (#19653)
697bfe1d000d is described below
commit 697bfe1d000d22346489986b5f9b0c02be81c8fa
Author: Aditya Nikam <[email protected]>
AuthorDate: Wed Aug 19 08:39:01 2026 +0530
fix(flink): sort ClientIds heartbeat files numerically instead of
lexicographically (#19653)
---
.../main/java/org/apache/hudi/util/ClientIds.java | 10 ++-
.../java/org/apache/hudi/util/TestClientIds.java | 76 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 1 deletion(-)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
index c2b1c1d3a14f..a998c4358a2f 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/util/ClientIds.java
@@ -186,7 +186,7 @@ public class ClientIds implements AutoCloseable,
Serializable {
}
List<Path> sortedPaths =
Arrays.stream(fs.listStatus(heartbeatFolderPath))
.map(FileStatus::getPath)
- .sorted(Comparator.comparing(Path::getName))
+ .sorted(Comparator.comparingInt(path ->
getClientIdSortKey(getClientId(path))))
.collect(Collectors.toList());
if (sortedPaths.isEmpty()) {
return INIT_CLIENT_ID;
@@ -219,6 +219,14 @@ public class ClientIds implements AutoCloseable,
Serializable {
return splits.length > 1 ? splits[1] : INIT_CLIENT_ID;
}
+ /**
+ * Returns a sort key so heartbeat files order numerically instead of
lexicographically.
+ * The base heartbeat file (empty client id) always sorts first.
+ */
+ private static int getClientIdSortKey(String clientId) {
+ return StringUtils.isNullOrEmpty(clientId) ? -1 :
Integer.parseInt(clientId);
+ }
+
// -------------------------------------------------------------------------
// Inner classes
// -------------------------------------------------------------------------
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestClientIds.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestClientIds.java
new file mode 100644
index 000000000000..5ea1dfb8e8a0
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/util/TestClientIds.java
@@ -0,0 +1,76 @@
+/*
+ * 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.util;
+
+import org.apache.hudi.configuration.FlinkOptions;
+
+import org.apache.flink.configuration.Configuration;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for {@link ClientIds}.
+ */
+public class TestClientIds {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ public void testNextIdSortsNumericallyAcrossDoubleDigitIds() throws
IOException {
+ createHeartbeatFiles("", "1", "2", "3", "4", "5", "6", "7", "8", "9",
"10");
+ Configuration conf = confForBasePath();
+
+ String nextId = ClientIds.builder().conf(conf).build().nextId(conf);
+
+ assertEquals("11", nextId);
+ }
+
+ @Test
+ public void testNextIdSingleDigitIdsStillWork() throws IOException {
+ createHeartbeatFiles("", "1", "2");
+ Configuration conf = confForBasePath();
+
+ String nextId = ClientIds.builder().conf(conf).build().nextId(conf);
+
+ assertEquals("3", nextId);
+ }
+
+ private void createHeartbeatFiles(String... clientIds) throws IOException {
+ Path heartbeatDir =
tempDir.resolve(".hoodie").resolve(".aux").resolve(".ids");
+ Files.createDirectories(heartbeatDir);
+ for (String clientId : clientIds) {
+ Files.createFile(heartbeatDir.resolve("_" + clientId));
+ }
+ }
+
+ private Configuration confForBasePath() {
+ String uri = tempDir.toUri().toString();
+ String basePath = uri.endsWith("/") ? uri.substring(0, uri.length() - 1) :
uri;
+ Configuration conf = new Configuration();
+ conf.set(FlinkOptions.PATH, basePath);
+ return conf;
+ }
+}
\ No newline at end of file
