Skip to content

fix(timeline-service): fail marker creation when create request fails #19369

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nsivabalan wants to merge 1 commit into
base: master
Choose a base branch
from
+59 −2
Open

fix(timeline-service): fail marker creation when create request fails #19369

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions ...nt-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,10 @@
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.exception.HoodieEarlyConflictDetectionException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieRemoteException;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
Expand Down Expand Up @@ -137,7 +139,7 @@ protected Option<StoragePath> create(String partitionPath, String fileName, IOTy
if (success) {
return Option.of(new StoragePath(FSUtils.constructAbsolutePath(markerDirPath, partitionPath), markerFileName));
} else {
return Option.empty();
throw new HoodieIOException("[timeline-server-based] Failed to create marker for partition " + partitionPath + ", fileName " + fileName + " with IOType " + type);

@wombatu-kun wombatu-kun Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A false result here does not mean the create failed: the server returns false when the marker already exists (MarkerDirState sets isSuccessful to !exists), and a genuine I/O failure already surfaces as HoodieRemoteException, not as false. So this throw fires on an already-existing marker, which createIfNotExists/createLogMarkerIfNotExists treat as a harmless Option.empty no-op - is turning already-exists into a hard failure intended, or should the throw be limited to a real create failure?

@nsivabalan nsivabalan Jul 24, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hey @wombatu-kun : thanks for the note.
none of our code path should try to create an already existing marker. every file is immutable including log files. and task retries will use a diff write token. so, a client can never ask for to create a marker that already exists.

let me know if I am missing any case here

@hudi-agent hudi-agent Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 One case worth double-checking here: createIfNotExists / createLogMarkerIfNotExists route through this same overridden create(..., checkIfExists=true), and this timeline-server impl ignores checkIfExists. So the server returning false for an already-existing marker (MarkerDirState#processMarkerCreationRequests does future.setIsSuccessful(!exists)) now throws for those callers too, where it previously returned a benign Option.empty().

A couple of real callers depend on that no-op rather than a fresh write token: HoodieWriteHandle#getLogCreationCallback calls createLogMarkerIfNotExists(...).isPresent() (so an existing APPEND log marker is expected to yield empty, not an exception), and RollbackHelperV1 uses createIfNotExists for APPEND markers during rollback retries. Since genuine I/O failures already surface as HoodieRemoteException, would it be safer to scope the throw to the checkIfExists == false path (or have the server distinguish already-exists from a real create failure) so the *IfNotExists contract is preserved?

@hudi-agent hudi-agent Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 This create(...) override ignores its checkIfExists param, so it now throws on the createIfNotExists path too. The server returns success=false both on a real failure and when the marker already exists, whereas DirectWriteMarkers.create() returns Option.empty() for the already-exists case. Callers that rely on that idempotency — createLogMarkerIfNotExists in HoodieWriteHandle, and Flink's FlinkCreateHandle/FlinkMergeHandle.createMarkerFile — would now get an exception instead of an empty Option. Could we gate the throw on !checkIfExists (or otherwise distinguish already-exists from failure) so the two marker impls stay consistent? @nsivabalan

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

}
}

Expand Down Expand Up @@ -168,7 +170,8 @@ public Option<StoragePath> createWithEarlyConflictDetection(String partitionPath
* @param markerFileName Marker file name.
* @return {@code true} if successful; {@code false} otherwise.
*/
private boolean executeCreateMarkerRequest(Map<String, String> paramsMap, String partitionPath, String markerFileName) {
@VisibleForTesting
boolean executeCreateMarkerRequest(Map<String, String> paramsMap, String partitionPath, String markerFileName) {
boolean success;
try {
success = executeRequestToTimelineServer(
Expand Down
54 changes: 54 additions & 0 deletions ...lient/src/test/java/org/apache/hudi/table/marker/TestTimelineServerBasedWriteMarkers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,14 +23,17 @@
import org.apache.hudi.common.config.HoodieMetadataConfig;
import org.apache.hudi.common.engine.HoodieEngineContext;
import org.apache.hudi.common.engine.HoodieLocalEngineContext;
import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.table.marker.MarkerType;
import org.apache.hudi.common.table.view.FileSystemViewManager;
import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
import org.apache.hudi.common.table.view.FileSystemViewStorageType;
import org.apache.hudi.common.util.MarkerUtils;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieRemoteException;
import org.apache.hudi.io.util.FileIOUtils;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.table.HoodieTable;
import org.apache.hudi.testutils.HoodieClientTestUtils;
import org.apache.hudi.timeline.service.TimelineService;
import org.apache.hudi.timeline.service.TimelineServiceTestHarness;
Expand All @@ -40,6 +43,7 @@
import org.apache.spark.api.java.JavaSparkContext;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.function.Executable;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.EnumSource;
Expand All @@ -49,13 +53,15 @@
import java.io.InputStream;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static org.apache.hudi.common.table.view.FileSystemViewStorageType.SPILLABLE_DISK;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertIterableEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

@Slf4j
public class TestTimelineServerBasedWriteMarkers extends TestWriteMarkersBase {
Expand Down Expand Up @@ -162,6 +168,54 @@ private void restartServerAndClient(int numberOfSimulatedConnectionFailures,
}
}

@Test
public void testMarkerCreationFailure() throws IOException {
FileSystemViewStorageConfig.Builder builder = FileSystemViewStorageConfig.newBuilder().withRemoteServerHost("localhost")
.withRemoteServerPort(timelineService.getServerPort())
.withRemoteTimelineClientTimeoutSecs(DEFAULT_READ_TIMEOUT_SECS);
MockTimelineServerBasedWriteMarkers timelineServerBasedWriteMarkers = new MockTimelineServerBasedWriteMarkers(basePath, markerFolderPath.toString(), "000", builder.build());
// this should succeed.
timelineServerBasedWriteMarkers.create("2020/06/01", "file1", IOType.MERGE);

assertTrue(storage.exists(markerFolderPath));
assertTrue(writeMarkers.doesMarkerDirExist());

// lets fail the marker creation
timelineServerBasedWriteMarkers.failMarkerCreation = true;
try {
timelineServerBasedWriteMarkers.create("2020/06/01", "file2", IOType.MERGE);
fail("Should not have reached here");
} catch (HoodieIOException ioe) {
assertTrue(ioe.getMessage().contains("[timeline-server-based] Failed to create marker for partition"));
} finally {

@wombatu-kun wombatu-kun Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@AfterEach cleanup() already closes timelineService, so closing it again in this finally double-closes the service. Drop the finally - the try only needs to wrap the failing create.

if (timelineService != null) {
timelineService.close();
}
}
}

static class MockTimelineServerBasedWriteMarkers extends TimelineServerBasedWriteMarkers {

boolean failMarkerCreation = false;

@hudi-agent hudi-agent Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 nit: failMarkerCreation reads as an imperative verb rather than a boolean predicate — could you rename it to something like shouldFailMarkerCreation so that if (!shouldFailMarkerCreation) reads more naturally?

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.


public MockTimelineServerBasedWriteMarkers(HoodieTable table, String instantTime) {

@wombatu-kun wombatu-kun Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This MockTimelineServerBasedWriteMarkers(HoodieTable, String) constructor is unused - the test builds the mock through the (String, String, String, FileSystemViewStorageConfig) constructor. Remove it.

super(table, instantTime);
}

MockTimelineServerBasedWriteMarkers(String basePath, String markerFolderPath, String instantTime, FileSystemViewStorageConfig fileSystemViewStorageConfig) {
super(basePath, markerFolderPath, instantTime, fileSystemViewStorageConfig);
}

@Override
boolean executeCreateMarkerRequest(Map<String, String> paramsMap, String partitionPath, String markerFileName) {
if (!failMarkerCreation) {
return super.executeCreateMarkerRequest(paramsMap, partitionPath, markerFileName);

@hudi-agent hudi-agent Jul 27, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🤖 nit: the else branch here is redundant since the if block always returns — dropping it to a bare return false; after the closing brace would be slightly cleaner.

⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.

} else {
return false;

@wombatu-kun wombatu-kun Jul 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Returning false here mirrors a create failure, but the real server also returns false when the marker already exists (MarkerDirState: !exists), so the test cannot tell a genuine failure from a benign already-exists and never covers createIfNotExists on an existing marker. Add a case that creates the same marker twice to lock down the intended behavior.

}
}
}

private static TimelineServerBasedWriteMarkers initWriteMarkers(String basePath,
String markerFolderPath,
int serverPort,
Expand Down
Loading