This is an automated email from the ASF dual-hosted git repository.
nsivabalan 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 ebbfcc500dc1 fix(timeline-service): fail marker creation requests when
marker flush fails (#19368)
ebbfcc500dc1 is described below
commit ebbfcc500dc138fb32d33e2778d6ba3ed0b88c37
Author: Y Ethan Guo <[email protected]>
AuthorDate: Thu Jul 23 21:27:15 2026 -0700
fix(timeline-service): fail marker creation requests when marker flush
fails (#19368)
MarkerDirState.flushMarkersToFile closed the marker file stream with
closeQuietly, so a close()-time flush failure (e.g. S3A, which performs
the object PUT in close()) was silently swallowed. The pending marker
creation request was then acknowledged as successful, letting the
writer create a data file with no durable marker. Marker-based
rollback then missed such files, and a retry under the same instant
could leave duplicate file groups behind.
- Close the writer within try-with-resources so a close() failure
propagates as HoodieIOException instead of being swallowed.
- On a flush failure, roll back the markers buffered by the failed
batch from in-memory state and fail the pending futures exceptionally,
so a retried request can recreate the markers.
- Release the marker file index in a finally block so a failure
doesn't leak it.
Adds TestMarkerDirState covering the success, flush-failure, and
retry-after-failure paths.
---
.../service/handlers/marker/MarkerDirState.java | 122 +++++++++------
.../handlers/marker/TestMarkerDirState.java | 170 +++++++++++++++++++++
2 files changed, 248 insertions(+), 44 deletions(-)
diff --git
a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
index 9f804086f481..715a8171f6c4 100644
---
a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
+++
b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
@@ -40,7 +40,6 @@ import org.apache.hadoop.util.StringUtils;
import java.io.BufferedWriter;
import java.io.IOException;
-import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
@@ -54,7 +53,6 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.hudi.common.util.MarkerUtils.MARKERS_FILENAME_PREFIX;
-import static org.apache.hudi.io.util.FileIOUtils.closeQuietly;
import static org.apache.hudi.timeline.service.RequestHandler.jsonifyResult;
/**
@@ -207,46 +205,62 @@ public class MarkerDirState implements Serializable {
log.debug("timeMs={} markerDirPath={} numRequests={} fileIndex={}",
System.currentTimeMillis(), markerDirPath,
pendingMarkerCreationFutures.size(), fileIndex);
boolean shouldFlushMarkers = false;
-
- synchronized (markerCreationProcessingLock) {
- for (MarkerCreationFuture future : pendingMarkerCreationFutures) {
- String markerName = future.getMarkerName();
- boolean exists = allMarkers.contains(markerName);
- if (!exists) {
- if (conflictDetectionStrategy.isPresent()) {
- try {
-
conflictDetectionStrategy.get().detectAndResolveConflictIfNecessary();
- } catch (HoodieEarlyConflictDetectionException he) {
- log.error("Detected the write conflict due to a concurrent
writer, "
- + "failing the marker creation as the early conflict
detection is enabled", he);
- future.setIsSuccessful(false);
- continue;
- } catch (Exception e) {
- log.warn("Failed to execute early conflict detection. Marker
creation will continue.", e);
- // When early conflict detection fails to execute, we still
allow the marker creation
- // to continue
- addMarkerToMap(fileIndex, markerName);
- future.setIsSuccessful(true);
- shouldFlushMarkers = true;
- continue;
+ int fileMarkersLengthBeforeBatch = 0;
+
+ try {
+ synchronized (markerCreationProcessingLock) {
+ StringBuilder fileMarkers = fileMarkersMap.get(fileIndex);
+ fileMarkersLengthBeforeBatch = fileMarkers == null ? 0 :
fileMarkers.length();
+ for (MarkerCreationFuture future : pendingMarkerCreationFutures) {
+ String markerName = future.getMarkerName();
+ boolean exists = allMarkers.contains(markerName);
+ if (!exists) {
+ if (conflictDetectionStrategy.isPresent()) {
+ try {
+
conflictDetectionStrategy.get().detectAndResolveConflictIfNecessary();
+ } catch (HoodieEarlyConflictDetectionException he) {
+ log.error("Detected the write conflict due to a concurrent
writer, "
+ + "failing the marker creation as the early conflict
detection is enabled", he);
+ future.setIsSuccessful(false);
+ continue;
+ } catch (Exception e) {
+ log.warn("Failed to execute early conflict detection. Marker
creation will continue.", e);
+ // When early conflict detection fails to execute, we still
allow the marker creation
+ // to continue
+ addMarkerToMap(fileIndex, markerName);
+ future.setIsSuccessful(true);
+ shouldFlushMarkers = true;
+ continue;
+ }
}
+ addMarkerToMap(fileIndex, markerName);
+ shouldFlushMarkers = true;
}
- addMarkerToMap(fileIndex, markerName);
- shouldFlushMarkers = true;
+ future.setIsSuccessful(!exists);
}
- future.setIsSuccessful(!exists);
- }
- if (!isMarkerTypeWritten) {
- // Create marker directory and write marker type to MARKERS.type
- writeMarkerTypeToFile();
- isMarkerTypeWritten = true;
+ if (!isMarkerTypeWritten) {
+ // Create marker directory and write marker type to MARKERS.type
+ writeMarkerTypeToFile();
+ isMarkerTypeWritten = true;
+ }
}
+ if (shouldFlushMarkers) {
+ flushMarkersToFile(fileIndex);
+ }
+ } catch (Exception e) {
+ log.error("Failed to persist markers to file index {} in {}", fileIndex,
markerDirPath, e);
+ // The markers added by this batch are not durably persisted, so they
are removed from
+ // the in-memory state and all pending requests fail, so that no write
operation
+ // proceeds without a durable marker and a retried request can recreate
the marker
+ removeMarkersOfPendingFutures(pendingMarkerCreationFutures, fileIndex,
fileMarkersLengthBeforeBatch);
+ for (MarkerCreationFuture future : pendingMarkerCreationFutures) {
+ future.completeExceptionally(e);
+ }
+ return;
+ } finally {
+ markFileAsAvailable(fileIndex);
}
- if (shouldFlushMarkers) {
- flushMarkersToFile(fileIndex);
- }
- markFileAsAvailable(fileIndex);
for (MarkerCreationFuture future : pendingMarkerCreationFutures) {
try {
@@ -309,6 +323,29 @@ public class MarkerDirState implements Serializable {
stringBuilder.append('\n');
}
+ /**
+ * Removes the markers added by the pending marker creation requests from
the in-memory state,
+ * used when the markers cannot be persisted, so that a retried request can
recreate the markers.
+ *
+ * @param pendingMarkerCreationFutures futures of pending marker creation
requests
+ * @param fileIndex file index used by the batch of
requests
+ * @param fileMarkersLengthBeforeBatch length of the buffered markers of the
file index before the batch
+ */
+ private void removeMarkersOfPendingFutures(
+ List<MarkerCreationFuture> pendingMarkerCreationFutures, int fileIndex,
int fileMarkersLengthBeforeBatch) {
+ synchronized (markerCreationProcessingLock) {
+ for (MarkerCreationFuture future : pendingMarkerCreationFutures) {
+ if (future.isSuccessful()) {
+ allMarkers.remove(future.getMarkerName());
+ }
+ }
+ StringBuilder fileMarkers = fileMarkersMap.get(fileIndex);
+ if (fileMarkers != null) {
+ fileMarkers.setLength(fileMarkersLengthBeforeBatch);
+ }
+ }
+ }
+
/**
* Writes marker type, "TIMELINE_SERVER_BASED", to file.
*/
@@ -357,17 +394,14 @@ public class MarkerDirState implements Serializable {
HoodieTimer timer = HoodieTimer.start();
StoragePath markersFilePath = new StoragePath(
markerDirPath, MARKERS_FILENAME_PREFIX + markerFileIndex);
- OutputStream outputStream = null;
- BufferedWriter bufferedWriter = null;
- try {
- outputStream = storage.create(markersFilePath);
- bufferedWriter = new BufferedWriter(new OutputStreamWriter(outputStream,
StandardCharsets.UTF_8));
+ // The writer must be closed within the try scope, so that a failure to
persist the markers
+ // at close() time, e.g., when an object store uploads the file content in
close(), is
+ // propagated to the caller instead of being swallowed
+ try (BufferedWriter bufferedWriter = new BufferedWriter(new
OutputStreamWriter(
+ storage.create(markersFilePath), StandardCharsets.UTF_8))) {
bufferedWriter.write(fileMarkersMap.get(markerFileIndex).toString());
} catch (IOException e) {
throw new HoodieIOException("Failed to overwrite marker file " +
markersFilePath, e);
- } finally {
- closeQuietly(bufferedWriter);
- closeQuietly(outputStream);
}
log.debug("{} written in {} ms", markersFilePath, timer.endTimer());
}
diff --git
a/hudi-timeline-service/src/test/java/org/apache/hudi/timeline/service/handlers/marker/TestMarkerDirState.java
b/hudi-timeline-service/src/test/java/org/apache/hudi/timeline/service/handlers/marker/TestMarkerDirState.java
new file mode 100644
index 000000000000..76597609c290
--- /dev/null
+++
b/hudi-timeline-service/src/test/java/org/apache/hudi/timeline/service/handlers/marker/TestMarkerDirState.java
@@ -0,0 +1,170 @@
+/*
+ * 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.timeline.service.handlers.marker;
+
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.io.util.FileIOUtils;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
+
+import io.javalin.http.Context;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.hudi.common.util.MarkerUtils.MARKERS_FILENAME_PREFIX;
+import static org.apache.hudi.common.util.MarkerUtils.MARKER_TYPE_FILENAME;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Tests {@link MarkerDirState}.
+ */
+public class TestMarkerDirState extends HoodieCommonTestHarness {
+ private static final String MARKER_NAME =
+
"2016/68b3ad4d-9d43-4e4c-8f4b-4c9b6d3e8b3a-0_1-0-1_00000000000001.parquet.marker.CREATE";
+
+ private CloseFailingHoodieStorage storage;
+ private String markerDir;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ initPath();
+ storage = new CloseFailingHoodieStorage(basePath, new Configuration());
+ markerDir = basePath + "/.hoodie/.temp/00000000000001";
+ }
+
+ @Test
+ void testMarkerCreationRequestProcessing() throws Exception {
+ MarkerDirState dirState = createMarkerDirState();
+ MarkerCreationFuture future = createFuture();
+ int fileIndex = dirState.getNextFileIndexToUse().get();
+
+ dirState.processMarkerCreationRequests(Collections.singletonList(future),
fileIndex);
+
+ assertTrue(future.isSuccessful());
+ assertEquals("true", future.get());
+ assertEquals(Collections.singleton(MARKER_NAME), dirState.getAllMarkers());
+ assertEquals(MARKER_NAME + "\n", readMarkersFileContent(fileIndex));
+ // The file index must be released after the batch is processed
+ assertEquals(Option.of(fileIndex), dirState.getNextFileIndexToUse());
+ }
+
+ @Test
+ void testMarkerCreationRequestsFailWhenFlushFails() {
+ MarkerDirState dirState = createMarkerDirState();
+ storage.setShouldFailClose(true);
+ MarkerCreationFuture future = createFuture();
+ int fileIndex = dirState.getNextFileIndexToUse().get();
+
+ dirState.processMarkerCreationRequests(Collections.singletonList(future),
fileIndex);
+
+ assertTrue(future.isCompletedExceptionally());
+ ExecutionException exception = assertThrows(ExecutionException.class,
future::get);
+ assertInstanceOf(HoodieIOException.class, exception.getCause());
+ // The marker without durable persistence must not be acknowledged as
existing
+ assertTrue(dirState.getAllMarkers().isEmpty());
+ // The file index must be released even if the flush fails
+ assertEquals(Option.of(fileIndex), dirState.getNextFileIndexToUse());
+ }
+
+ @Test
+ void testMarkerRecreationAfterFlushFailure() throws Exception {
+ MarkerDirState dirState = createMarkerDirState();
+ storage.setShouldFailClose(true);
+ MarkerCreationFuture failedFuture = createFuture();
+
dirState.processMarkerCreationRequests(Collections.singletonList(failedFuture),
0);
+ assertTrue(failedFuture.isCompletedExceptionally());
+
+ // A retried request for the same marker must succeed once the storage
recovers
+ storage.setShouldFailClose(false);
+ MarkerCreationFuture retriedFuture = createFuture();
+
dirState.processMarkerCreationRequests(Collections.singletonList(retriedFuture),
0);
+
+ assertTrue(retriedFuture.isSuccessful());
+ assertEquals("true", retriedFuture.get());
+ assertEquals(Collections.singleton(MARKER_NAME), dirState.getAllMarkers());
+ assertEquals(MARKER_NAME + "\n", readMarkersFileContent(0));
+ }
+
+ private MarkerDirState createMarkerDirState() {
+ return new MarkerDirState(
+ markerDir, 1, Option.empty(), storage,
Registry.getRegistry("TestMarkerDirState"), 1);
+ }
+
+ private MarkerCreationFuture createFuture() {
+ return new MarkerCreationFuture(mock(Context.class), markerDir,
MARKER_NAME);
+ }
+
+ private String readMarkersFileContent(int fileIndex) throws IOException {
+ return FileIOUtils.readAsUTFString(
+ storage.open(new StoragePath(markerDir, MARKERS_FILENAME_PREFIX +
fileIndex)));
+ }
+
+ /**
+ * Storage whose created streams for {@code MARKERS} index files fail at
close() time,
+ * simulating an object store that uploads the file content in close().
+ */
+ private static class CloseFailingHoodieStorage extends HoodieHadoopStorage {
+ private boolean shouldFailClose = false;
+
+ CloseFailingHoodieStorage(String path, Configuration conf) {
+ super(path, conf);
+ }
+
+ void setShouldFailClose(boolean shouldFailClose) {
+ this.shouldFailClose = shouldFailClose;
+ }
+
+ @Override
+ public OutputStream create(StoragePath path) throws IOException {
+ OutputStream stream = super.create(path);
+ String fileName = path.getName();
+ if (shouldFailClose
+ && fileName.startsWith(MARKERS_FILENAME_PREFIX) &&
!fileName.equals(MARKER_TYPE_FILENAME)) {
+ return new CloseFailingStream(stream);
+ }
+ return stream;
+ }
+ }
+
+ private static class CloseFailingStream extends FilterOutputStream {
+ CloseFailingStream(OutputStream delegate) {
+ super(delegate);
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ throw new IOException("Simulated failure to persist the file at close()
time");
+ }
+ }
+}
