nsivabalan commented on a change in pull request #3233: URL: https://github.com/apache/hudi/pull/3233#discussion_r680407670
########## File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/DirectMarkerFiles.java ########## @@ -175,70 +165,30 @@ public static String stripMarkerSuffix(String path) { return markerFiles; } - private String stripMarkerFolderPrefix(String fullMarkerPath) { - ValidationUtils.checkArgument(fullMarkerPath.contains(HoodieTableMetaClient.MARKER_EXTN)); - String markerRootPath = Path.getPathWithoutSchemeAndAuthority( - new Path(String.format("%s/%s/%s", basePath, HoodieTableMetaClient.TEMPFOLDER_NAME, instantTime))).toString(); - int begin = fullMarkerPath.indexOf(markerRootPath); - ValidationUtils.checkArgument(begin >= 0, - "Not in marker dir. Marker Path=" + fullMarkerPath + ", Expected Marker Root=" + markerRootPath); - return fullMarkerPath.substring(begin + markerRootPath.length() + 1); - } - - /** - * The marker path will be <base-path>/.hoodie/.temp/<instant_ts>/2019/04/25/filename.marker.writeIOType. - */ - public Path create(String partitionPath, String dataFileName, IOType type) { + @Override + protected Option<Path> create(String partitionPath, String dataFileName, IOType type, boolean checkIfExists) { + HoodieTimer timer = new HoodieTimer().startTimer(); Path markerPath = getMarkerPath(partitionPath, dataFileName, type); + Path dirPath = markerPath.getParent(); try { - LOG.info("Creating Marker Path=" + markerPath); - fs.create(markerPath, false).close(); + if (!fs.exists(dirPath)) { Review comment: not sure if this is addressed. can you help me understand how this is same as prior to this patch ########## 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: fix variable names to markerType. ########## File path: hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java ########## @@ -304,7 +304,7 @@ protected void postCommit(HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, Option<Map<String, String>> extraMetadata) { try { // Delete the marker directory for the instant. - new MarkerFiles(createTable(config, hadoopConf), instantTime) + MarkerFilesFactory.get(config.getMarkersType(), createTable(config, hadoopConf), instantTime) Review comment: So, does this new marker strategy work for all engines (spark, flink and java) w/o any additional changes? If not, can you file a follow up ticket. ########## 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: guess you are still fixing the docs. just adding a reminder. ########## 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: We can never get a repeat create marker request for a given file is it? and so we fail such calls? ########## 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: rename to TestMarkerFilesBase ########## 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: lazyInitComplete ########## 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: lazyInitLock ########## 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: You can probably make this Atomicboolean. And we can do something like if (firstMarkerCreationRequest.getAndSet(false)) { executorService.execute(markerCreationDispatchingRunnable); } ########## 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); + } + } + flushMarkersToFile(fileIndex); + + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + return pendingMarkerCreationFutures; + } + + /** + * Deletes markers in the directory. + * + * @return {@code true} if successful; {@code false} otherwise. + */ + public boolean deleteAllMarkers() { + Path dirPath = new Path(markerDirPath); + boolean result = false; + try { + if (fileSystem.exists(dirPath)) { + FileStatus[] fileStatuses = fileSystem.listStatus(dirPath); + List<String> markerDirSubPaths = Arrays.stream(fileStatuses) + .map(fileStatus -> fileStatus.getPath().toString()) + .collect(Collectors.toList()); + + if (markerDirSubPaths.size() > 0) { + SerializableConfiguration conf = new SerializableConfiguration(fileSystem.getConf()); + int actualParallelism = Math.min(markerDirSubPaths.size(), parallelism); + hoodieEngineContext.foreach(markerDirSubPaths, subPathStr -> { + Path subPath = new Path(subPathStr); + FileSystem fileSystem = subPath.getFileSystem(conf.get()); + fileSystem.delete(subPath, true); + }, actualParallelism); + } + + result = fileSystem.delete(dirPath, true); + LOG.info("Removing marker directory at " + dirPath); + } + } catch (IOException ioe) { + throw new HoodieIOException(ioe.getMessage(), ioe); + } + allMarkers.clear(); + fileMarkersMap.clear(); + return result; + } + + /** + * Syncs the markers from the underlying files for the first request. + */ + private void maybeSyncOnFirstRequest() { + synchronized (firstRequestLock) { + if (!hasFirstRequest) { + syncMarkersFromFileSystem(); + hasFirstRequest = true; + } + } + } + + /** + * @return the next file index to use in a round-robin fashion + */ + private int getNextFileIndexToUse() { + int fileIndex = -1; + synchronized (threadUseStatus) { + for (int i = 0; i < threadUseStatus.size(); i++) { Review comment: for loop is bit confusing me. Ideally if PUT request suceeds within 1 sec, we should never go in a loop. First index after last known index should be available to server. but here is my suggestion. Will leave it to you to take a call. We can first check if (lastFileIndex + 1 ) is free to use. If not, we can go into a for loop. Easier to read and reason about. ########## 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: why call sync again here? I see that fetchPendingMarkerCreationRequests() is calling it anyway. Is there a scenario where processMarkerCreationRequests will be called w/o fetchPendingMarkerCreationRequests? -- 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: commits-unsubscr...@hudi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org