diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
index e763a21d33777..3665f7ddc775a 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/TimelineServerBasedWriteMarkers.java
@@ -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;
@@ -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);
     }
   }
 
@@ -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(
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestTimelineServerBasedWriteMarkers.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestTimelineServerBasedWriteMarkers.java
index 82a8254569b6c..1dc605846db2d 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestTimelineServerBasedWriteMarkers.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestTimelineServerBasedWriteMarkers.java
@@ -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;
@@ -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;
@@ -49,6 +53,7 @@
 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;
@@ -56,6 +61,7 @@
 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 {
@@ -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 {
+      if (timelineService != null) {
+        timelineService.close();
+      }
+    }
+  }
+
+  static class MockTimelineServerBasedWriteMarkers extends TimelineServerBasedWriteMarkers {
+
+    boolean failMarkerCreation = false;
+
+    public MockTimelineServerBasedWriteMarkers(HoodieTable table, String instantTime) {
+      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);
+      } else {
+        return false;
+      }
+    }
+  }
+
   private static TimelineServerBasedWriteMarkers initWriteMarkers(String basePath,
                                                                   String markerFolderPath,
                                                                   int serverPort,