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 55becc7b03f4 fix: do not fall back to timeline server markers on
transient HDFS failures (#18887)
55becc7b03f4 is described below
commit 55becc7b03f4cf8ec3e1e5303d14fd46bfc8f9b9
Author: vamshipasunuru1 <[email protected]>
AuthorDate: Wed Jul 22 01:53:52 2026 -0700
fix: do not fall back to timeline server markers on transient HDFS failures
(#18887)
* fix: do not fall back to timeline server markers on transient HDFS
failures
When MARKERS.type is absent, MarkerBasedRollbackUtils tries DIRECT markers
and catches IOException to fall back to TIMELINE_SERVER_BASED. This was too
broad: a transient "Server too busy" RetriableException is also an
IOException,
causing rollback to use the timeline server marker path which finds 0
markers
and deletes nothing, leaving orphan data files behind.
Only catch IllegalArgumentException (marker path format mismatch) for the
fallback. Let IOException propagate so the rollback fails and retries rather
than silently producing an incorrect result.
* update log
* Address review: use parameterized SLF4J logging and fix storage reference
- Switch String.format to SLF4J parameterized form so the exception
stack trace is preserved in the log output.
- Fix compile error: getTimelineServerBasedMarkers takes HoodieStorage
(storage), not the undefined fileSystem variable.
* Address review: rewrite test to exercise the actual code path
The prior test stubbed metaClient.getFs() and
HoodieWrapperFileSystem.listStatus(),
but the production code goes through table.getStorage() and
HoodieStorage.listDirectEntries().
The injected IOException never reached the catch block under test.
Rewrite to mock the correct seams:
- table.getStorage() returns a mock HoodieStorage
- storage.exists(MARKERS.type) → false to trigger the fallback branch
- storage.exists(markerDir) → true so allMarkerFilePaths() lists entries
- storage.listDirectEntries(markerDir) throws the failure of interest
Verified by inspecting the surefire stack trace: the exception now
originates
in FSUtils.processFiles → DirectWriteMarkers.allMarkerFilePaths →
MarkerBasedRollbackUtils.getAllMarkerPaths, confirming the fix is exercised.
Also add a companion test asserting that IllegalArgumentException still
falls
back to TIMELINE_SERVER_BASED (preserving the original intended behavior).
---------
Co-authored-by: vamshi_UBER <[email protected]>
Co-authored-by: vamshipasunuru1 <[email protected]>
---
.../table/marker/MarkerBasedRollbackUtils.java | 15 ++-
.../table/marker/TestMarkerBasedRollbackUtils.java | 134 +++++++++++++++++++++
2 files changed, 146 insertions(+), 3 deletions(-)
diff --git
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/MarkerBasedRollbackUtils.java
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/MarkerBasedRollbackUtils.java
index e7e539fa8899..562f6bc9221b 100644
---
a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/MarkerBasedRollbackUtils.java
+++
b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/MarkerBasedRollbackUtils.java
@@ -70,9 +70,18 @@ public class MarkerBasedRollbackUtils {
WriteMarkers writeMarkers = WriteMarkersFactory.get(DIRECT, table,
instant);
try {
return new ArrayList<>(writeMarkers.allMarkerFilePaths());
- } catch (IOException | IllegalArgumentException e) {
- log.warn("{} not present and {} marker failed with error: {}. Falling
back to {} marker",
- MARKER_TYPE_FILENAME, DIRECT, e.getMessage(),
TIMELINE_SERVER_BASED);
+ } catch (IOException e) {
+ // Do NOT fall back to TIMELINE_SERVER_BASED on transient IO failures
(e.g., HDFS throttling).
+ // The timeline server looks in a different location and would return
0 markers, causing the
+ // rollback to skip deleting data files and leaving orphan files on
the table.
+ log.warn("{} not present and {} marker listing failed with IO error. "
+ + "Propagating exception, rollback will retry rather than fall
back to {}.",
+ MARKER_TYPE_FILENAME, DIRECT, TIMELINE_SERVER_BASED, e);
+ throw e;
+ } catch (IllegalArgumentException e) {
+ // IllegalArgumentException indicates a marker path format mismatch,
fall back to timeline server.
+ log.warn("{} not present and {} marker failed. Falling back to {}
marker",
+ MARKER_TYPE_FILENAME, DIRECT, TIMELINE_SERVER_BASED, e);
return getTimelineServerBasedMarkers(context, parallelism, markerDir,
storage);
}
}
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/marker/TestMarkerBasedRollbackUtils.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/marker/TestMarkerBasedRollbackUtils.java
new file mode 100644
index 000000000000..cc964e4c65d3
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/table/marker/TestMarkerBasedRollbackUtils.java
@@ -0,0 +1,134 @@
+/*
+ * 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.table.marker;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.table.HoodieTable;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.apache.hudi.common.util.MarkerUtils.MARKER_TYPE_FILENAME;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link MarkerBasedRollbackUtils}.
+ *
+ * <p>These tests target the {@code MARKERS.type}-absent code path in
+ * {@link MarkerBasedRollbackUtils#getAllMarkerPaths}. That branch attempts to
list DIRECT
+ * markers first and used to catch {@code IOException |
IllegalArgumentException} and fall
+ * back to TIMELINE_SERVER_BASED. A transient HDFS failure would therefore be
swallowed and
+ * the rollback would return zero marker paths, leaving orphan data files on
the table.
+ *
+ * <p>The tests mock the {@link HoodieStorage} seams the production code
actually goes
+ * through:
+ * <ul>
+ * <li>{@code readMarkerType(...)} -> {@code
storage.exists(MARKERS.type)} = false
+ * <li>{@code DirectWriteMarkers.doesMarkerDirExist()} -> {@code
storage.exists(markerDir)} = true
+ * <li>{@code FSUtils.processFiles} -> {@code
storage.listDirectEntries(markerDir)} throws
+ * </ul>
+ */
+public class TestMarkerBasedRollbackUtils {
+
+ private static final String INSTANT = "20260101000000";
+ private static final String BASE_PATH = "/tmp/test-table";
+ private static final String MARKER_DIR = BASE_PATH + "/.hoodie/.temp/" +
INSTANT;
+
+ private HoodieTable mockTable;
+ private HoodieTableMetaClient mockMetaClient;
+ private HoodieEngineContext mockContext;
+ private HoodieStorage mockStorage;
+
+ @BeforeEach
+ public void setUp() throws IOException {
+ mockTable = mock(HoodieTable.class);
+ mockMetaClient = mock(HoodieTableMetaClient.class);
+ mockContext = mock(HoodieEngineContext.class);
+ mockStorage = mock(HoodieStorage.class);
+ HoodieTableConfig mockTableConfig = mock(HoodieTableConfig.class);
+
+ when(mockTable.getMetaClient()).thenReturn(mockMetaClient);
+ when(mockTable.getContext()).thenReturn(mockContext);
+ when(mockTable.getStorage()).thenReturn(mockStorage);
+ when(mockMetaClient.getBasePath()).thenReturn(new StoragePath(BASE_PATH));
+ when(mockMetaClient.getMarkerFolderPath(INSTANT)).thenReturn(MARKER_DIR);
+ when(mockMetaClient.getTableConfig()).thenReturn(mockTableConfig);
+ // Table version 8+ selects DirectWriteMarkers in WriteMarkersFactory.
+
when(mockTableConfig.getTableVersion()).thenReturn(HoodieTableVersion.EIGHT);
+
+ StoragePath markerDirPath = new StoragePath(MARKER_DIR);
+ StoragePath markerTypeFilePath = new StoragePath(markerDirPath,
MARKER_TYPE_FILENAME);
+ // MARKERS.type is absent, this drives execution into the fallback branch
under test.
+ when(mockStorage.exists(markerTypeFilePath)).thenReturn(false);
+ // The marker directory itself exists so
DirectWriteMarkers.allMarkerFilePaths()
+ // proceeds to list entries (rather than early-returning an empty set).
+ when(mockStorage.exists(markerDirPath)).thenReturn(true);
+ }
+
+ /**
+ * A transient IO failure while listing DIRECT markers must propagate as
IOException
+ * rather than silently falling back to TIMELINE_SERVER_BASED. Falling back
would return
+ * zero marker paths (timeline server uses a different location) and cause
rollback to
+ * leave orphan data files on the table.
+ */
+ @Test
+ public void
testGetAllMarkerPathsPropagatesIOExceptionOnTransientListingFailure() throws
IOException {
+ when(mockStorage.listDirectEntries(new StoragePath(MARKER_DIR)))
+ .thenThrow(new IOException("Server too busy - disconnecting"));
+
+ IOException thrown = assertThrows(IOException.class,
+ () -> MarkerBasedRollbackUtils.getAllMarkerPaths(mockTable,
mockContext, INSTANT, 1));
+ // Verify the original transient error surfaces to the caller (not a
wrapped fallback error).
+ assertEquals("Server too busy - disconnecting", thrown.getMessage());
+ }
+
+ /**
+ * IllegalArgumentException (e.g., marker path format mismatch) must retain
the original
+ * fallback behavior, read markers via TIMELINE_SERVER_BASED path instead of
failing.
+ *
+ * <p>Both DirectWriteMarkers and the timeline-server-based reader end up
calling
+ * {@code storage.listDirectEntries(markerDir)}. We stub the first
invocation to throw
+ * (triggering the fallback under test) and subsequent invocations to return
an empty
+ * list so the fallback path completes cleanly.
+ */
+ @Test
+ public void
testGetAllMarkerPathsFallsBackToTimelineServerOnIllegalArgumentException()
throws IOException {
+ when(mockStorage.listDirectEntries(new StoragePath(MARKER_DIR)))
+ .thenThrow(new IllegalArgumentException("bad marker path"))
+ .thenReturn(Collections.emptyList());
+
+ // No exception should surface, the IllegalArgumentException must be
handled by the
+ // fallback rather than propagating to the caller.
+ assertDoesNotThrow(
+ () -> MarkerBasedRollbackUtils.getAllMarkerPaths(mockTable,
mockContext, INSTANT, 1));
+ }
+}
