[
https://issues.apache.org/jira/browse/HUDI-1138?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17391456#comment-17391456
]
ASF GitHub Bot commented on HUDI-1138:
--------------------------------------
yihua commented on a change in pull request #3233:
URL: https://github.com/apache/hudi/pull/3233#discussion_r680737342
##########
File path:
hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestMarkerFiles.java
##########
@@ -25,49 +25,27 @@
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.testutils.HoodieClientTestUtils;
-import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
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 java.io.IOException;
-import java.util.List;
import java.util.stream.Collectors;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertIterableEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
-public class TestMarkerFiles extends HoodieCommonTestHarness {
-
- private MarkerFiles markerFiles;
- private FileSystem fs;
- private Path markerFolderPath;
- private JavaSparkContext jsc;
- private HoodieSparkEngineContext context;
-
- @BeforeEach
- public void setup() throws IOException {
- initPath();
- initMetaClient();
- this.jsc = new
JavaSparkContext(HoodieClientTestUtils.getSparkConfForTest(TestMarkerFiles.class.getName()));
- this.context = new HoodieSparkEngineContext(jsc);
- this.fs = FSUtils.getFs(metaClient.getBasePath(),
metaClient.getHadoopConf());
- this.markerFolderPath = new Path(metaClient.getMarkerFolderPath("000"));
- this.markerFiles = new MarkerFiles(fs, metaClient.getBasePath(),
markerFolderPath.toString(), "000");
- }
+public abstract class TestMarkerFiles extends HoodieCommonTestHarness {
Review comment:
Fixed.
##########
File path:
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.config.SerializableConfiguration;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.FileIOUtils.closeQuietly;
+
+/**
+ * Stores the state of a marker directory.
+ *
+ * The operations inside this class is designed to be thread-safe.
+ */
+public class MarkerDirState implements Serializable {
+ public static final String MARKERS_FILENAME_PREFIX = "MARKERS";
+ private static final Logger LOG = LogManager.getLogger(MarkerDirState.class);
+ // Marker directory
+ private final String markerDirPath;
+ private final FileSystem fileSystem;
+ // A cached copy of all markers in memory
+ private final Set<String> allMarkers = new HashSet<>();
+ // A cached copy of marker entries in each marker file, stored in
StringBuilder
+ // for efficient appending
+ // Mapping: {markerFileIndex -> markers}
+ private final Map<Integer, StringBuilder> fileMarkersMap = new HashMap<>();
+ // A list of use status of underlying files storing markers by a thread.
+ // {@code true} means the file is in use by a {@code
BatchCreateMarkerRunnable}.
+ // Index of the list is used for the filename, i.e., "1" -> "MARKERS1"
+ private final List<Boolean> threadUseStatus;
+ // A list of pending futures from async marker creation requests
+ private final List<MarkerCreationCompletableFuture> markerCreationFutures =
new ArrayList<>();
+ private final int parallelism;
+ private final Object firstRequestLock = new Object();
+ private final Object markerCreationProcessingLock = new Object();
+ private transient HoodieEngineContext hoodieEngineContext;
+ // Last underlying file index used, for finding the next file index
+ // in a round-robin fashion
+ private int lastFileIndex = 0;
+ private boolean hasFirstRequest = false;
+
+ public MarkerDirState(String markerDirPath, int markerBatchNumThreads,
+ FileSystem fileSystem, HoodieEngineContext
hoodieEngineContext, int parallelism) {
+ this.markerDirPath = markerDirPath;
+ this.fileSystem = fileSystem;
+ this.hoodieEngineContext = hoodieEngineContext;
+ this.parallelism = parallelism;
+
+ List<Boolean> isMarkerFileInUseList = new
ArrayList<>(markerBatchNumThreads);
+ for (int i = 0; i < markerBatchNumThreads; i++) {
+ isMarkerFileInUseList.add(false);
+ }
+ this.threadUseStatus = Collections.synchronizedList(isMarkerFileInUseList);
+ }
+
+ /**
+ * @return {@code true} if the marker directory exists in the system.
+ */
+ public boolean exists() {
+ try {
+ return fileSystem.exists(new Path(markerDirPath));
+ } catch (IOException ioe) {
+ throw new HoodieIOException(ioe.getMessage(), ioe);
+ }
+ }
+
+ /**
+ * @return all markers in the marker directory.
+ */
+ public Set<String> getAllMarkers() {
+ maybeSyncOnFirstRequest();
+ return allMarkers;
+ }
+
+ /**
+ * Adds a {@code MarkerCreationCompletableFuture} instance from a marker
+ * creation request to the queue.
+ *
+ * @param future {@code MarkerCreationCompletableFuture} instance.
+ */
+ public void addMarkerCreationFuture(MarkerCreationCompletableFuture future) {
+ synchronized (markerCreationFutures) {
+ markerCreationFutures.add(future);
+ }
+ }
+
+ /**
+ * @return futures of pending marker creation requests.
+ */
+ public List<MarkerCreationCompletableFuture>
fetchPendingMarkerCreationRequests() {
+ if (markerCreationFutures.isEmpty()) {
+ return new ArrayList<>();
+ }
+ maybeSyncOnFirstRequest();
+ List<MarkerCreationCompletableFuture> pendingFutures;
+ synchronized (markerCreationFutures) {
+ pendingFutures = new ArrayList<>(markerCreationFutures);
+ markerCreationFutures.clear();
+ }
+ return pendingFutures;
+ }
+
+ /**
+ * Processes pending marker creation requests if possible.
+ *
+ * @param pendingMarkerCreationFutures futures of pending marker creation
requests
+ * @return A list of processed futures
+ */
+ public List<MarkerCreationCompletableFuture> processMarkerCreationRequests(
+ final List<MarkerCreationCompletableFuture>
pendingMarkerCreationFutures) {
+ if (pendingMarkerCreationFutures.isEmpty()) {
+ return Collections.emptyList();
+ }
+ maybeSyncOnFirstRequest();
+
+ int fileIndex = getNextFileIndexToUse();
+ if (fileIndex < 0) {
+ LOG.debug("All marker files are busy, skip batch processing of create
marker requests in " + markerDirPath);
+ synchronized (markerCreationFutures) {
+ markerCreationFutures.addAll(pendingMarkerCreationFutures);
+ }
+ return Collections.emptyList();
+ }
+ LOG.debug("timeMs=" + System.currentTimeMillis() + " fileIndex=" +
fileIndex);
+
+ synchronized (markerCreationProcessingLock) {
+ for (MarkerCreationCompletableFuture future :
pendingMarkerCreationFutures) {
+ String markerName = future.getMarkerName();
+ boolean exists = allMarkers.contains(markerName);
+ if (!exists) {
+ allMarkers.add(markerName);
+ StringBuilder stringBuilder =
fileMarkersMap.computeIfAbsent(fileIndex, k -> new StringBuilder(16384));
+ stringBuilder.append(markerName);
+ stringBuilder.append('\n');
+ }
+ future.setResult(!exists);
Review comment:
There could be two marker creation requests with the same marker name.
If that happens, the second request would still get a response from the
timeline server, with a boolean result of `false` indicating that the marker
has already been created.
##########
File path:
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java
##########
@@ -227,6 +228,30 @@
+ "files from lake storage, before committing the write. Reduce this
value, if the high number of tasks incur delays for smaller tables "
+ "or low latency writes.");
+ public static final ConfigProperty<String> MARKERS_TYPE_PROP = ConfigProperty
+ .key("hoodie.write.markers.type")
+ .defaultValue(MarkerType.DIRECT.toString())
+ .sinceVersion("0.9.0")
+ .withDocumentation("Marker IO mode to use. Two modes are supported: "
Review comment:
Right. Fixed.
##########
File path:
hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/upgrade/ZeroToOneUpgradeHandler.java
##########
@@ -78,6 +80,7 @@ public void upgrade(HoodieWriteConfig config,
HoodieEngineContext context, Strin
private static void recreateMarkerFiles(final String commitInstantTime,
HoodieSparkTable table,
HoodieEngineContext context,
+ MarkerType ioMode,
Review comment:
Fixed.
##########
File path:
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
##########
@@ -0,0 +1,180 @@
+/*
+ * 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;
+
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.model.IOType;
+import org.apache.hudi.common.table.view.FileSystemViewManager;
+import org.apache.hudi.timeline.service.TimelineService;
+import
org.apache.hudi.timeline.service.handlers.marker.MarkerCreationCompletableFuture;
+import
org.apache.hudi.timeline.service.handlers.marker.MarkerCreationDispatchingRunnable;
+import org.apache.hudi.timeline.service.handlers.marker.MarkerDirState;
+
+import io.javalin.Context;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
+
+/**
+ * REST Handler servicing marker requests.
+ *
+ * The marker creation requests are handled asynchronous, while other types of
requests
+ * are handled synchronous.
+ *
+ * Marker creation requests are batch processed periodically by a thread.
Each batch
+ * processing thread adds new markers to a marker file. Given that marker
file operation
+ * can take time, multiple concurrent threads can run at the same, while they
operate
+ * on different marker files storing mutually exclusive marker entries. At
any given
+ * time, a marker file is touched by at most one thread to guarantee
consistency.
+ * Below is an example of running batch processing threads.
+ *
+ * |-----| batch interval
+ * Thread 1 |-------------------------->| writing to MARKERS1
+ * Thread 2 |-------------------------->| writing to MARKERS2
+ * Thread 3 |-------------------------->| writing to MARKERS3
+ */
+public class MarkerHandler extends Handler {
+ private static final Logger LOG = LogManager.getLogger(MarkerHandler.class);
+
+ private final ExecutorService executorService;
+ // Parallelism for reading and deleting marker files
+ private final int parallelism;
+ // Marker directory states, {markerDirPath -> MarkerDirState instance}
+ private final Map<String, MarkerDirState> markerDirStateMap = new
HashMap<>();
+ // A long-running thread to dispatch marker creation requests to batch
processing threads
+ private final MarkerCreationDispatchingRunnable
markerCreationDispatchingRunnable;
+ private transient HoodieEngineContext hoodieEngineContext;
+ // Lock for synchronous processing of marker creating requests
+ private final Object createMarkerRequestLockObject = new Object();
+ // Next batch process timestamp in milliseconds
+ private boolean firstMarkerCreationRequest = true;
+
+ public MarkerHandler(Configuration conf, TimelineService.Config
timelineServiceConfig,
+ HoodieEngineContext hoodieEngineContext, FileSystem
fileSystem,
+ FileSystemViewManager viewManager, Registry
metricsRegistry) throws IOException {
+ super(conf, timelineServiceConfig, fileSystem, viewManager);
+ LOG.debug("MarkerHandler FileSystem: " + this.fileSystem.getScheme());
+ LOG.debug("MarkerHandler batching params: batchNumThreads=" +
timelineServiceConfig.markerBatchNumThreads
+ + " batchIntervalMs=" + timelineServiceConfig.markerBatchIntervalMs +
"ms");
+ this.hoodieEngineContext = hoodieEngineContext;
+ this.parallelism = timelineServiceConfig.markerParallelism;
+ this.executorService = Executors.newSingleThreadExecutor();
+ this.markerCreationDispatchingRunnable = new
MarkerCreationDispatchingRunnable(
+ markerDirStateMap, metricsRegistry,
timelineServiceConfig.markerBatchNumThreads,
+ timelineServiceConfig.markerBatchIntervalMs);
+ }
+
+ public void stop() {
+ markerCreationDispatchingRunnable.stop();
+ }
+
+ /**
+ * @param markerDir marker directory path
+ * @return all marker paths in the marker directory
+ */
+ public Set<String> getAllMarkers(String markerDir) {
+ MarkerDirState markerDirState =
markerDirStateMap.computeIfAbsent(markerDir,
+ k -> new MarkerDirState(markerDir,
timelineServiceConfig.markerBatchNumThreads, fileSystem,
+ hoodieEngineContext, parallelism));
+ return markerDirState.getAllMarkers();
+ }
+
+ /**
+ * @param markerDir marker directory path
+ * @return all marker paths of write IO type "CREATE" and "MERGE"
+ */
+ public Set<String> getCreateAndMergeMarkers(String markerDir) {
+ return getAllMarkers(markerDir).stream()
+ .filter(markerName -> !markerName.endsWith(IOType.APPEND.name()))
+ .collect(Collectors.toSet());
+ }
+
+ /**
+ * @param markerDir marker directory path
+ * @return {@code true} if the marker directory exists; {@code false}
otherwise.
+ */
+ public boolean doesMarkerDirExist(String markerDir) {
+ MarkerDirState markerDirState =
markerDirStateMap.computeIfAbsent(markerDir,
+ k -> new MarkerDirState(markerDir,
timelineServiceConfig.markerBatchNumThreads, fileSystem,
+ hoodieEngineContext, parallelism));
+ return markerDirState.exists();
+ }
+
+ /**
+ * Generates a future for an async marker creation request
+ *
+ * The future is added to the marker creation future list and waits for the
next batch processing
+ * of marker creation requests.
+ *
+ * @param context Javalin app context
+ * @param markerDir marker directory path
+ * @param markerName marker name
+ * @return the {@code CompletableFuture} instance for the request
+ */
+ public CompletableFuture<String> createMarker(Context context, String
markerDir, String markerName) {
+ LOG.info("Request: create marker " + markerDir + " " + markerName);
+ MarkerCreationCompletableFuture future = new
MarkerCreationCompletableFuture(context, markerDir, markerName);
+ // Add the future to the list
+ MarkerDirState markerDirState = markerDirStateMap.get(markerDir);
+ if (markerDirState == null) {
+ synchronized (markerDirStateMap) {
+ markerDirState = new MarkerDirState(markerDir,
timelineServiceConfig.markerBatchNumThreads,
+ fileSystem, hoodieEngineContext, parallelism);
+ markerDirStateMap.put(markerDir, markerDirState);
+ }
+ }
+ markerDirState.addMarkerCreationFuture(future);
+ synchronized (createMarkerRequestLockObject) {
+ if (firstMarkerCreationRequest) {
Review comment:
Right, that's better. Addressed.
##########
File path:
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.config.SerializableConfiguration;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.FileIOUtils.closeQuietly;
+
+/**
+ * Stores the state of a marker directory.
+ *
+ * The operations inside this class is designed to be thread-safe.
+ */
+public class MarkerDirState implements Serializable {
+ public static final String MARKERS_FILENAME_PREFIX = "MARKERS";
+ private static final Logger LOG = LogManager.getLogger(MarkerDirState.class);
+ // Marker directory
+ private final String markerDirPath;
+ private final FileSystem fileSystem;
+ // A cached copy of all markers in memory
+ private final Set<String> allMarkers = new HashSet<>();
+ // A cached copy of marker entries in each marker file, stored in
StringBuilder
+ // for efficient appending
+ // Mapping: {markerFileIndex -> markers}
+ private final Map<Integer, StringBuilder> fileMarkersMap = new HashMap<>();
+ // A list of use status of underlying files storing markers by a thread.
+ // {@code true} means the file is in use by a {@code
BatchCreateMarkerRunnable}.
+ // Index of the list is used for the filename, i.e., "1" -> "MARKERS1"
+ private final List<Boolean> threadUseStatus;
+ // A list of pending futures from async marker creation requests
+ private final List<MarkerCreationCompletableFuture> markerCreationFutures =
new ArrayList<>();
+ private final int parallelism;
+ private final Object firstRequestLock = new Object();
+ private final Object markerCreationProcessingLock = new Object();
+ private transient HoodieEngineContext hoodieEngineContext;
+ // Last underlying file index used, for finding the next file index
+ // in a round-robin fashion
+ private int lastFileIndex = 0;
+ private boolean hasFirstRequest = false;
Review comment:
Fixed.
##########
File path:
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
##########
@@ -0,0 +1,383 @@
+/*
+ * 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.config.SerializableConfiguration;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.metrics.Registry;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.exception.HoodieException;
+import org.apache.hudi.exception.HoodieIOException;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.hudi.common.util.FileIOUtils.closeQuietly;
+import static org.apache.hudi.timeline.service.RequestHandler.jsonifyResult;
+
+/**
+ * Stores the state of a marker directory.
+ *
+ * The operations inside this class is designed to be thread-safe.
+ */
+public class MarkerDirState implements Serializable {
+ public static final String MARKERS_FILENAME_PREFIX = "MARKERS";
+ private static final Logger LOG = LogManager.getLogger(MarkerDirState.class);
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+ // Marker directory
+ private final String markerDirPath;
+ private final FileSystem fileSystem;
+ private final Registry metricsRegistry;
+ // A cached copy of all markers in memory
+ private final Set<String> allMarkers = new HashSet<>();
+ // A cached copy of marker entries in each marker file, stored in
StringBuilder
+ // for efficient appending
+ // Mapping: {markerFileIndex -> markers}
+ private final Map<Integer, StringBuilder> fileMarkersMap = new HashMap<>();
+ // A list of use status of underlying files storing markers by a thread.
+ // {@code true} means the file is in use by a {@code
BatchCreateMarkerRunnable}.
+ // Index of the list is used for the filename, i.e., "1" -> "MARKERS1"
+ private final List<Boolean> threadUseStatus;
+ // A list of pending futures from async marker creation requests
+ private final List<MarkerCreationCompletableFuture> markerCreationFutures =
new ArrayList<>();
+ private final int parallelism;
+ private final AtomicBoolean lazyInitComplete;
+ private final Object markerCreationProcessingLock = new Object();
+ private transient HoodieEngineContext hoodieEngineContext;
+ // Last underlying file index used, for finding the next file index
+ // in a round-robin fashion
+ private int lastFileIndex = 0;
+
+ public MarkerDirState(String markerDirPath, int markerBatchNumThreads,
FileSystem fileSystem,
+ Registry metricsRegistry, HoodieEngineContext
hoodieEngineContext, int parallelism) {
+ this.markerDirPath = markerDirPath;
+ this.fileSystem = fileSystem;
+ this.metricsRegistry = metricsRegistry;
+ this.hoodieEngineContext = hoodieEngineContext;
+ this.parallelism = parallelism;
+ this.threadUseStatus =
+ Stream.generate(() ->
false).limit(markerBatchNumThreads).collect(Collectors.toList());
+ this.lazyInitComplete = new AtomicBoolean(false);
+ }
+
+ /**
+ * @return {@code true} if the marker directory exists in the system.
+ */
+ public boolean exists() {
+ try {
+ return fileSystem.exists(new Path(markerDirPath));
+ } catch (IOException ioe) {
+ throw new HoodieIOException(ioe.getMessage(), ioe);
+ }
+ }
+
+ /**
+ * @return all markers in the marker directory.
+ */
+ public Set<String> getAllMarkers() {
+ maybeSyncOnFirstRequest();
+ return allMarkers;
+ }
+
+ /**
+ * Adds a {@code MarkerCreationCompletableFuture} instance from a marker
+ * creation request to the queue.
+ *
+ * @param future {@code MarkerCreationCompletableFuture} instance.
+ */
+ public void addMarkerCreationFuture(MarkerCreationCompletableFuture future) {
+ synchronized (markerCreationFutures) {
+ markerCreationFutures.add(future);
+ }
+ }
+
+ /**
+ * @return the next file index to use in a round-robin fashion,
+ * or -1 if no file is available.
+ */
+ public int getNextFileIndexToUse() {
+ int fileIndex = -1;
+ synchronized (threadUseStatus) {
+ for (int i = 0; i < threadUseStatus.size(); i++) {
+ int index = (lastFileIndex + 1 + i) % threadUseStatus.size();
Review comment:
@nsivabalan the logic has already checked the next index after the last
used index, so if the next index is available for use, the loop ends there.
##########
File path:
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.config.SerializableConfiguration;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.FileIOUtils.closeQuietly;
+
+/**
+ * Stores the state of a marker directory.
+ *
+ * The operations inside this class is designed to be thread-safe.
+ */
+public class MarkerDirState implements Serializable {
+ public static final String MARKERS_FILENAME_PREFIX = "MARKERS";
+ private static final Logger LOG = LogManager.getLogger(MarkerDirState.class);
+ // Marker directory
+ private final String markerDirPath;
+ private final FileSystem fileSystem;
+ // A cached copy of all markers in memory
+ private final Set<String> allMarkers = new HashSet<>();
+ // A cached copy of marker entries in each marker file, stored in
StringBuilder
+ // for efficient appending
+ // Mapping: {markerFileIndex -> markers}
+ private final Map<Integer, StringBuilder> fileMarkersMap = new HashMap<>();
+ // A list of use status of underlying files storing markers by a thread.
+ // {@code true} means the file is in use by a {@code
BatchCreateMarkerRunnable}.
+ // Index of the list is used for the filename, i.e., "1" -> "MARKERS1"
+ private final List<Boolean> threadUseStatus;
+ // A list of pending futures from async marker creation requests
+ private final List<MarkerCreationCompletableFuture> markerCreationFutures =
new ArrayList<>();
+ private final int parallelism;
+ private final Object firstRequestLock = new Object();
Review comment:
I used AtomicBoolean here as suggested in another place.
##########
File path:
hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java
##########
@@ -0,0 +1,371 @@
+/*
+ * 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.config.SerializableConfiguration;
+import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.util.HoodieTimer;
+import org.apache.hudi.common.util.collection.ImmutablePair;
+import org.apache.hudi.exception.HoodieIOException;
+
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.util.StringUtils;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.util.FileIOUtils.closeQuietly;
+
+/**
+ * Stores the state of a marker directory.
+ *
+ * The operations inside this class is designed to be thread-safe.
+ */
+public class MarkerDirState implements Serializable {
+ public static final String MARKERS_FILENAME_PREFIX = "MARKERS";
+ private static final Logger LOG = LogManager.getLogger(MarkerDirState.class);
+ // Marker directory
+ private final String markerDirPath;
+ private final FileSystem fileSystem;
+ // A cached copy of all markers in memory
+ private final Set<String> allMarkers = new HashSet<>();
+ // A cached copy of marker entries in each marker file, stored in
StringBuilder
+ // for efficient appending
+ // Mapping: {markerFileIndex -> markers}
+ private final Map<Integer, StringBuilder> fileMarkersMap = new HashMap<>();
+ // A list of use status of underlying files storing markers by a thread.
+ // {@code true} means the file is in use by a {@code
BatchCreateMarkerRunnable}.
+ // Index of the list is used for the filename, i.e., "1" -> "MARKERS1"
+ private final List<Boolean> threadUseStatus;
+ // A list of pending futures from async marker creation requests
+ private final List<MarkerCreationCompletableFuture> markerCreationFutures =
new ArrayList<>();
+ private final int parallelism;
+ private final Object firstRequestLock = new Object();
+ private final Object markerCreationProcessingLock = new Object();
+ private transient HoodieEngineContext hoodieEngineContext;
+ // Last underlying file index used, for finding the next file index
+ // in a round-robin fashion
+ private int lastFileIndex = 0;
+ private boolean hasFirstRequest = false;
+
+ public MarkerDirState(String markerDirPath, int markerBatchNumThreads,
+ FileSystem fileSystem, HoodieEngineContext
hoodieEngineContext, int parallelism) {
+ this.markerDirPath = markerDirPath;
+ this.fileSystem = fileSystem;
+ this.hoodieEngineContext = hoodieEngineContext;
+ this.parallelism = parallelism;
+
+ List<Boolean> isMarkerFileInUseList = new
ArrayList<>(markerBatchNumThreads);
+ for (int i = 0; i < markerBatchNumThreads; i++) {
+ isMarkerFileInUseList.add(false);
+ }
+ this.threadUseStatus = Collections.synchronizedList(isMarkerFileInUseList);
+ }
+
+ /**
+ * @return {@code true} if the marker directory exists in the system.
+ */
+ public boolean exists() {
+ try {
+ return fileSystem.exists(new Path(markerDirPath));
+ } catch (IOException ioe) {
+ throw new HoodieIOException(ioe.getMessage(), ioe);
+ }
+ }
+
+ /**
+ * @return all markers in the marker directory.
+ */
+ public Set<String> getAllMarkers() {
+ maybeSyncOnFirstRequest();
+ return allMarkers;
+ }
+
+ /**
+ * Adds a {@code MarkerCreationCompletableFuture} instance from a marker
+ * creation request to the queue.
+ *
+ * @param future {@code MarkerCreationCompletableFuture} instance.
+ */
+ public void addMarkerCreationFuture(MarkerCreationCompletableFuture future) {
+ synchronized (markerCreationFutures) {
+ markerCreationFutures.add(future);
+ }
+ }
+
+ /**
+ * @return futures of pending marker creation requests.
+ */
+ public List<MarkerCreationCompletableFuture>
fetchPendingMarkerCreationRequests() {
+ if (markerCreationFutures.isEmpty()) {
+ return new ArrayList<>();
+ }
+ maybeSyncOnFirstRequest();
+ List<MarkerCreationCompletableFuture> pendingFutures;
+ synchronized (markerCreationFutures) {
+ pendingFutures = new ArrayList<>(markerCreationFutures);
+ markerCreationFutures.clear();
+ }
+ return pendingFutures;
+ }
+
+ /**
+ * Processes pending marker creation requests if possible.
+ *
+ * @param pendingMarkerCreationFutures futures of pending marker creation
requests
+ * @return A list of processed futures
+ */
+ public List<MarkerCreationCompletableFuture> processMarkerCreationRequests(
+ final List<MarkerCreationCompletableFuture>
pendingMarkerCreationFutures) {
+ if (pendingMarkerCreationFutures.isEmpty()) {
+ return Collections.emptyList();
+ }
+ maybeSyncOnFirstRequest();
Review comment:
Right. There is duplication here.
I remove the `maybeSyncOnFirstRequest()` calling in
`fetchPendingMarkerCreationRequests()` since the fetching is invoked in the
dispatching and that should not include any file system operations which is
time-consuming. I keep the `maybeSyncOnFirstRequest()` calling in
`processMarkerCreationRequests()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
> Re-implement marker files via timeline server
> ---------------------------------------------
>
> Key: HUDI-1138
> URL: https://issues.apache.org/jira/browse/HUDI-1138
> Project: Apache Hudi
> Issue Type: Improvement
> Components: Writer Core
> Affects Versions: 0.9.0
> Reporter: Vinoth Chandar
> Assignee: Ethan Guo
> Priority: Blocker
> Labels: pull-request-available
> Fix For: 0.9.0
>
>
> Even as you can argue that RFC-15/consolidated metadata, removes the need for
> deleting partial files written due to spark task failures/stage retries. It
> will still leave extra files inside the table (and users will pay for it
> every month) and we need the marker mechanism to be able to delete these
> partial files.
> Here we explore if we can improve the current marker file mechanism, that
> creates one marker file per data file written, by
> Delegating the createMarker() call to the driver/timeline server, and have it
> create marker metadata into a single file handle, that is flushed for
> durability guarantees
>
> P.S: I was tempted to think Spark listener mechanism can help us deal with
> failed tasks, but it has no guarantees. the writer job could die without
> deleting a partial file. i.e it can improve things, but cant provide
> guarantees
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
