yihua commented on a change in pull request #3233: URL: https://github.com/apache/hudi/pull/3233#discussion_r686407495
########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java ########## @@ -0,0 +1,193 @@ +/* + * 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.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +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 Registry metricsRegistry; + // a scheduled executor service to schedule dispatching of marker creation requests + private final ScheduledExecutorService dispatchingExecutorService; + // an executor service to schedule the worker threads of batch processing marker creation requests + private final ExecutorService batchingExecutorService; + // 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 thread to dispatch marker creation requests to batch processing threads + private final MarkerCreationDispatchingRunnable markerCreationDispatchingRunnable; + private final AtomicBoolean firstCreationRequestSeen; + private transient HoodieEngineContext hoodieEngineContext; + private ScheduledFuture<?> dispatchingScheduledFuture; Review comment: renamed to `dispatchingThreadFuture`. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java ########## @@ -0,0 +1,193 @@ +/* + * 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.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +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 Registry metricsRegistry; + // a scheduled executor service to schedule dispatching of marker creation requests + private final ScheduledExecutorService dispatchingExecutorService; + // an executor service to schedule the worker threads of batch processing marker creation requests + private final ExecutorService batchingExecutorService; + // 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 thread to dispatch marker creation requests to batch processing threads + private final MarkerCreationDispatchingRunnable markerCreationDispatchingRunnable; + private final AtomicBoolean firstCreationRequestSeen; + private transient HoodieEngineContext hoodieEngineContext; + private ScheduledFuture<?> dispatchingScheduledFuture; + + 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.metricsRegistry = metricsRegistry; + this.parallelism = timelineServiceConfig.markerParallelism; + this.dispatchingExecutorService = Executors.newSingleThreadScheduledExecutor(); + this.batchingExecutorService = Executors.newFixedThreadPool(timelineServiceConfig.markerBatchNumThreads); + this.markerCreationDispatchingRunnable = + new MarkerCreationDispatchingRunnable(markerDirStateMap, batchingExecutorService); + this.firstCreationRequestSeen = new AtomicBoolean(false); + } + + /** + * Stops the dispatching of marker creation requests. + */ + public void stop() { + if (dispatchingScheduledFuture != null) { + dispatchingScheduledFuture.cancel(true); + } + dispatchingExecutorService.shutdown(); + batchingExecutorService.shutdown(); + } + + /** + * @param markerDir marker directory path + * @return all marker paths in the marker directory + */ + public Set<String> getAllMarkers(String markerDir) { + MarkerDirState markerDirState = getMarkerDirState(markerDir); + 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 = getMarkerDirState(markerDir); + 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 = getMarkerDirState(markerDir); + markerDirState.addMarkerCreationFuture(future); + if (!firstCreationRequestSeen.getAndSet(true)) { + dispatchingScheduledFuture = dispatchingExecutorService.scheduleAtFixedRate(markerCreationDispatchingRunnable, Review comment: I assume the scheduling shouldn't fail here. To be safe, I changed the logic back to using a lock object and a boolean. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationCompletableFuture.java ########## @@ -0,0 +1,69 @@ +/* + * 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.util.HoodieTimer; + +import io.javalin.Context; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.util.concurrent.CompletableFuture; + +/** + * Future for async marker creation request. + */ +public class MarkerCreationCompletableFuture extends CompletableFuture<String> { + private static final Logger LOG = LogManager.getLogger(MarkerCreationCompletableFuture.class); + private final Context context; + private final String markerDirPath; + private final String markerName; + private boolean result; + private final HoodieTimer timer; + + public MarkerCreationCompletableFuture(Context context, String markerDirPath, String markerName) { + super(); + this.timer = new HoodieTimer().startTimer(); + this.context = context; + this.markerDirPath = markerDirPath; + this.markerName = markerName; + this.result = false; + } + + public Context getContext() { + return context; + } + + public String getMarkerDirPath() { + return markerDirPath; + } + + public String getMarkerName() { + return markerName; + } + + public boolean getResult() { Review comment: Fixed. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationBatchingRunnable.java ########## @@ -0,0 +1,51 @@ +/* + * 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.util.HoodieTimer; + +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.util.List; + +/** + * A runnable for batch processing marker creation requests. + */ +public class MarkerCreationBatchingRunnable implements Runnable { Review comment: Fixed. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + 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. + * + * @param pendingMarkerCreationFutures futures of pending marker creation requests + * @param fileIndex file index to use to write markers + */ + public void processMarkerCreationRequests( + final List<MarkerCreationCompletableFuture> pendingMarkerCreationFutures, int fileIndex) { + if (pendingMarkerCreationFutures.isEmpty()) { + markFileAsAvailable(fileIndex); + return; + } + + LOG.debug("timeMs=" + System.currentTimeMillis() + " markerDirPath=" + markerDirPath + + " numRequests=" + pendingMarkerCreationFutures.size() + " 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); + markFileAsAvailable(fileIndex); + + for (MarkerCreationCompletableFuture future : pendingMarkerCreationFutures) { + try { + future.complete(jsonifyResult( + future.getContext(), future.getResult(), metricsRegistry, OBJECT_MAPPER, LOG)); + } catch (JsonProcessingException e) { + throw new HoodieException("Failed to JSON encode the value", e); + } + } + } + + /** + * 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); Review comment: It's not necessary. Changed the second argument to false. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + 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. + * + * @param pendingMarkerCreationFutures futures of pending marker creation requests + * @param fileIndex file index to use to write markers + */ + public void processMarkerCreationRequests( + final List<MarkerCreationCompletableFuture> pendingMarkerCreationFutures, int fileIndex) { + if (pendingMarkerCreationFutures.isEmpty()) { + markFileAsAvailable(fileIndex); + return; + } + + LOG.debug("timeMs=" + System.currentTimeMillis() + " markerDirPath=" + markerDirPath + + " numRequests=" + pendingMarkerCreationFutures.size() + " 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); + markFileAsAvailable(fileIndex); + + for (MarkerCreationCompletableFuture future : pendingMarkerCreationFutures) { + try { + future.complete(jsonifyResult( + future.getContext(), future.getResult(), metricsRegistry, OBJECT_MAPPER, LOG)); + } catch (JsonProcessingException e) { + throw new HoodieException("Failed to JSON encode the value", e); + } + } + } + + /** + * 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(); Review comment: Based on the write process, for the same marker directory, there should not be any other requests when the deletion is happening, so we don't put a synchronization block here. ########## File path: hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/marker/MarkerFiles.java ########## @@ -0,0 +1,117 @@ +/* + * 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.table.marker; + +import org.apache.hudi.common.engine.HoodieEngineContext; +import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.IOType; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.util.ValidationUtils; +import org.apache.hudi.exception.HoodieIOException; + +import org.apache.hadoop.fs.Path; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.io.IOException; +import java.io.Serializable; +import java.util.Set; + +public abstract class MarkerFiles implements Serializable { Review comment: Missed this before. Yes, we can rename it, to make the naming general. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirRequestContext.java ########## @@ -0,0 +1,57 @@ +/* + * 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 java.util.List; + +/** + * Input of batch processing of marker creation requests for a single marker directory. + */ +public class MarkerDirRequestContext { Review comment: Right. Renamed. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationDispatchingRunnable.java ########## @@ -0,0 +1,87 @@ +/* + * 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.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; + +/** + * A runnable / worker for scheduling batch processing of marker creation requests. + */ +public class MarkerCreationDispatchingRunnable implements Runnable { + public static final Logger LOG = LogManager.getLogger(MarkerCreationDispatchingRunnable.class); + + // Marker directory states, {markerDirPath -> MarkerDirState instance} + private final Map<String, MarkerDirState> markerDirStateMap; + private final ExecutorService executorService; + + public MarkerCreationDispatchingRunnable( + Map<String, MarkerDirState> markerDirStateMap, ExecutorService executorService) { + this.markerDirStateMap = markerDirStateMap; + this.executorService = executorService; + } + + /** + * Dispatches the marker creation requests that can be process to a worker thread of batch + * processing the requests. + * + * For each marker directory, goes through the following steps: + * (1) find the next available file index for writing. If no file index is available, + * skip the processing of this marker directory; + * (2) fetch the pending marker creation requests for this marker directory. If there is + * no request, skip this marker directory; + * (3) put the marker directory, marker dir state, list of requests futures, and the file index + * to a {@code MarkerDirRequestContext} instance and add the instance to the request context list. + * + * If the request context list is not empty, spins up a worker thread, {@code MarkerCreationBatchingRunnable}, + * and pass all the request context to the thread for batch processing. The thread is responsible + * for responding to the request futures directly. + */ + @Override + public void run() { + List<MarkerDirRequestContext> requestContextList = new ArrayList<>(); + + // Only fetch pending marker creation requests that can be processed, + // i.e., that markers can be written to a underlying file + for (String markerDir : markerDirStateMap.keySet()) { + MarkerDirState markerDirState = markerDirStateMap.get(markerDir); + int fileIndex = markerDirState.getNextFileIndexToUse(); Review comment: Yes, that makes sense. Fixed. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationCompletableFuture.java ########## @@ -0,0 +1,69 @@ +/* + * 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.util.HoodieTimer; + +import io.javalin.Context; +import org.apache.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.util.concurrent.CompletableFuture; + +/** + * Future for async marker creation request. + */ +public class MarkerCreationCompletableFuture extends CompletableFuture<String> { Review comment: Fixed. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationDispatchingRunnable.java ########## @@ -0,0 +1,87 @@ +/* + * 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.log4j.LogManager; +import org.apache.log4j.Logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; + +/** + * A runnable / worker for scheduling batch processing of marker creation requests. Review comment: Rewritten. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + public List<MarkerCreationCompletableFuture> fetchPendingMarkerCreationRequests() { + if (markerCreationFutures.isEmpty()) { Review comment: Looks like `isEmpty()` is not thread-safe. Moved it into the synchronized block. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + public List<MarkerCreationCompletableFuture> fetchPendingMarkerCreationRequests() { + if (markerCreationFutures.isEmpty()) { + return new ArrayList<>(); + } + maybeSyncOnFirstRequest(); Review comment: Right. Putting the sync inside the constructor is better to avoid oversight. Fixed. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + 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. + * + * @param pendingMarkerCreationFutures futures of pending marker creation requests + * @param fileIndex file index to use to write markers + */ + public void processMarkerCreationRequests( + final List<MarkerCreationCompletableFuture> pendingMarkerCreationFutures, int fileIndex) { + if (pendingMarkerCreationFutures.isEmpty()) { + markFileAsAvailable(fileIndex); + return; + } + + LOG.debug("timeMs=" + System.currentTimeMillis() + " markerDirPath=" + markerDirPath + + " numRequests=" + pendingMarkerCreationFutures.size() + " 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); Review comment: This is fine for future requests as the marker name is unique. As long as the file write is successful later on, we are good. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { Review comment: Right. I decided to go with `i = 0` approach with a one-line comment to make the code concise and readable. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + 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. + * + * @param pendingMarkerCreationFutures futures of pending marker creation requests + * @param fileIndex file index to use to write markers + */ + public void processMarkerCreationRequests( + final List<MarkerCreationCompletableFuture> pendingMarkerCreationFutures, int fileIndex) { + if (pendingMarkerCreationFutures.isEmpty()) { + markFileAsAvailable(fileIndex); + return; + } + + LOG.debug("timeMs=" + System.currentTimeMillis() + " markerDirPath=" + markerDirPath + + " numRequests=" + pendingMarkerCreationFutures.size() + " 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); + markFileAsAvailable(fileIndex); + + for (MarkerCreationCompletableFuture future : pendingMarkerCreationFutures) { + try { + future.complete(jsonifyResult( + future.getContext(), future.getResult(), metricsRegistry, OBJECT_MAPPER, LOG)); + } catch (JsonProcessingException e) { + throw new HoodieException("Failed to JSON encode the value", e); + } + } + } + + /** + * 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 (lazyInitLock) { + if (!lazyInitComplete) { + syncMarkersFromFileSystem(); + lazyInitComplete = true; + } + } + } + + /** + * Syncs all markers maintained in the underlying files under the marker directory in the file system. + */ + private void syncMarkersFromFileSystem() { + Path dirPath = new Path(markerDirPath); + try { + if (fileSystem.exists(dirPath)) { + FileStatus[] fileStatuses = fileSystem.listStatus(dirPath); + List<String> markerDirSubPaths = Arrays.stream(fileStatuses) + .map(fileStatus -> fileStatus.getPath().toString()) + .filter(pathStr -> pathStr.contains(MARKERS_FILENAME_PREFIX)) + .collect(Collectors.toList()); + + if (markerDirSubPaths.size() > 0) { + SerializableConfiguration conf = new SerializableConfiguration(fileSystem.getConf()); + int actualParallelism = Math.min(markerDirSubPaths.size(), parallelism); + Map<String, Set<String>> fileMarkersSetMap = + hoodieEngineContext.mapToPair(markerDirSubPaths, markersFilePathStr -> { + Path markersFilePath = new Path(markersFilePathStr); + FileSystem fileSystem = markersFilePath.getFileSystem(conf.get()); + FSDataInputStream fsDataInputStream = null; + BufferedReader bufferedReader = null; + Set<String> markers = new HashSet<>(); + try { + LOG.debug("Read marker file: " + markersFilePathStr); + fsDataInputStream = fileSystem.open(markersFilePath); + bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream, StandardCharsets.UTF_8)); + markers = bufferedReader.lines().collect(Collectors.toSet()); + bufferedReader.close(); + fsDataInputStream.close(); + } catch (IOException e) { + throw new HoodieIOException("Failed to read MARKERS file " + markerDirPath, e); + } finally { + closeQuietly(bufferedReader); + closeQuietly(fsDataInputStream); + } + return new ImmutablePair<>(markersFilePathStr, markers); + }, actualParallelism); + + for (String markersFilePathStr: fileMarkersSetMap.keySet()) { + Set<String> fileMarkers = fileMarkersSetMap.get(markersFilePathStr); + if (!fileMarkers.isEmpty()) { + int index = parseMarkerFileIndex(markersFilePathStr); + + if (index >= 0) { + fileMarkersMap.put(index, new StringBuilder(StringUtils.join(",", fileMarkers))); + allMarkers.addAll(fileMarkers); + } + } + } + } + } + } catch (IOException ioe) { + throw new HoodieIOException(ioe.getMessage(), ioe); + } + } + + /** + * Parses the marker file index from the marker file path. + * + * E.g., if the marker file path is /tmp/table/.hoodie/.temp/000/MARKERS3, the index returned is 3. + * + * @param markerFilePathStr full path of marker file + * @return the marker file index + */ + private int parseMarkerFileIndex(String markerFilePathStr) { + String markerFileName = new Path(markerFilePathStr).getName(); + int prefixIndex = markerFileName.indexOf(MARKERS_FILENAME_PREFIX); + if (prefixIndex < 0) { + return -1; + } + try { + return Integer.parseInt(markerFileName.substring(prefixIndex + MARKERS_FILENAME_PREFIX.length())); + } catch (NumberFormatException nfe) { + LOG.error("Failed to parse marker file index from " + markerFilePathStr); + throw new HoodieException(nfe.getMessage(), nfe); + } + } + + /** + * Flushes markers to the underlying file. + * + * @param markerFileIndex file index to use. + */ + private void flushMarkersToFile(int markerFileIndex) { + LOG.debug("Write to " + markerDirPath + "/" + MARKERS_FILENAME_PREFIX + markerFileIndex); + HoodieTimer timer = new HoodieTimer().startTimer(); + Path markersFilePath = new Path(markerDirPath, MARKERS_FILENAME_PREFIX + markerFileIndex); + Path dirPath = markersFilePath.getParent(); + try { + if (!fileSystem.exists(dirPath)) { + fileSystem.mkdirs(dirPath); + } + } catch (IOException e) { + throw new HoodieIOException("Failed to make dir " + dirPath, e); + } + FSDataOutputStream fsDataOutputStream = null; + BufferedWriter bufferedWriter = null; + try { + fsDataOutputStream = fileSystem.create(markersFilePath); + bufferedWriter = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream, StandardCharsets.UTF_8)); + bufferedWriter.write(fileMarkersMap.get(markerFileIndex).toString()); Review comment: The failed overwrite on HDFS can be a problem. As discussed offline, we'll throw an exception for timeline-based markers on HDFS for now and fix the issue later. ########## File path: hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerDirState.java ########## @@ -0,0 +1,393 @@ +/* + * 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.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 Object lazyInitLock = 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 lastFileIndexUsed = 0; + private boolean lazyInitComplete; + + 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 = 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) { + int nextIndex = (lastFileIndexUsed + 1) % threadUseStatus.size(); + if (!threadUseStatus.get(nextIndex)) { + fileIndex = nextIndex; + threadUseStatus.set(nextIndex, true); + } else { + for (int i = 1; i < threadUseStatus.size(); i++) { + int index = (lastFileIndexUsed + 1 + i) % threadUseStatus.size(); + if (!threadUseStatus.get(index)) { + fileIndex = index; + threadUseStatus.set(index, true); + break; + } + } + } + if (fileIndex >= 0) { + lastFileIndexUsed = fileIndex; + } + } + return fileIndex; + } + + /** + * Marks the file as available to use again. + * + * @param fileIndex file index + */ + public void markFileAsAvailable(int fileIndex) { + synchronized (threadUseStatus) { + threadUseStatus.set(fileIndex, false); + } + } + + /** + * @return futures of pending marker creation requests and removes them from the list. + */ + 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. + * + * @param pendingMarkerCreationFutures futures of pending marker creation requests + * @param fileIndex file index to use to write markers + */ + public void processMarkerCreationRequests( + final List<MarkerCreationCompletableFuture> pendingMarkerCreationFutures, int fileIndex) { + if (pendingMarkerCreationFutures.isEmpty()) { + markFileAsAvailable(fileIndex); + return; + } + + LOG.debug("timeMs=" + System.currentTimeMillis() + " markerDirPath=" + markerDirPath + + " numRequests=" + pendingMarkerCreationFutures.size() + " 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); + markFileAsAvailable(fileIndex); + + for (MarkerCreationCompletableFuture future : pendingMarkerCreationFutures) { + try { + future.complete(jsonifyResult( + future.getContext(), future.getResult(), metricsRegistry, OBJECT_MAPPER, LOG)); + } catch (JsonProcessingException e) { + throw new HoodieException("Failed to JSON encode the value", e); + } + } + } + + /** + * 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 (lazyInitLock) { + if (!lazyInitComplete) { + syncMarkersFromFileSystem(); + lazyInitComplete = true; + } + } + } + + /** + * Syncs all markers maintained in the underlying files under the marker directory in the file system. + */ + private void syncMarkersFromFileSystem() { + Path dirPath = new Path(markerDirPath); + try { + if (fileSystem.exists(dirPath)) { + FileStatus[] fileStatuses = fileSystem.listStatus(dirPath); + List<String> markerDirSubPaths = Arrays.stream(fileStatuses) + .map(fileStatus -> fileStatus.getPath().toString()) + .filter(pathStr -> pathStr.contains(MARKERS_FILENAME_PREFIX)) + .collect(Collectors.toList()); + + if (markerDirSubPaths.size() > 0) { + SerializableConfiguration conf = new SerializableConfiguration(fileSystem.getConf()); + int actualParallelism = Math.min(markerDirSubPaths.size(), parallelism); + Map<String, Set<String>> fileMarkersSetMap = + hoodieEngineContext.mapToPair(markerDirSubPaths, markersFilePathStr -> { + Path markersFilePath = new Path(markersFilePathStr); + FileSystem fileSystem = markersFilePath.getFileSystem(conf.get()); + FSDataInputStream fsDataInputStream = null; + BufferedReader bufferedReader = null; + Set<String> markers = new HashSet<>(); + try { + LOG.debug("Read marker file: " + markersFilePathStr); + fsDataInputStream = fileSystem.open(markersFilePath); + bufferedReader = new BufferedReader(new InputStreamReader(fsDataInputStream, StandardCharsets.UTF_8)); + markers = bufferedReader.lines().collect(Collectors.toSet()); + bufferedReader.close(); + fsDataInputStream.close(); + } catch (IOException e) { + throw new HoodieIOException("Failed to read MARKERS file " + markerDirPath, e); + } finally { + closeQuietly(bufferedReader); + closeQuietly(fsDataInputStream); + } + return new ImmutablePair<>(markersFilePathStr, markers); + }, actualParallelism); + + for (String markersFilePathStr: fileMarkersSetMap.keySet()) { + Set<String> fileMarkers = fileMarkersSetMap.get(markersFilePathStr); + if (!fileMarkers.isEmpty()) { + int index = parseMarkerFileIndex(markersFilePathStr); + + if (index >= 0) { + fileMarkersMap.put(index, new StringBuilder(StringUtils.join(",", fileMarkers))); + allMarkers.addAll(fileMarkers); + } + } + } + } + } + } catch (IOException ioe) { + throw new HoodieIOException(ioe.getMessage(), ioe); + } + } + + /** + * Parses the marker file index from the marker file path. + * + * E.g., if the marker file path is /tmp/table/.hoodie/.temp/000/MARKERS3, the index returned is 3. + * + * @param markerFilePathStr full path of marker file + * @return the marker file index + */ + private int parseMarkerFileIndex(String markerFilePathStr) { Review comment: Yes. `syncMarkersFromFileSystem()` scans all the `MARKERS*` files under the marker directory, regardless of the `markerBatchNumThreads` set for a specific write. -- 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]
