reswqa commented on code in PR #22855:
URL: https://github.com/apache/flink/pull/22855#discussion_r1263431982
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileReader.java:
##########
@@ -42,8 +82,44 @@ public Buffer readBuffer(
MemorySegment memorySegment,
BufferRecycler recycler)
throws IOException {
- // TODO, implement the SegmentPartitionFileReader
- return null;
+
+ // Get the channel of the segment file for a subpartition.
+ Map<TieredStorageSubpartitionId, Tuple2<ReadableByteChannel, Integer>>
subpartitionInfo =
+ openedChannelAndSegmentIds.computeIfAbsent(partitionId, ignore
-> new HashMap<>());
+ Tuple2<ReadableByteChannel, Integer> fileChannelAndSegmentId =
+ subpartitionInfo.getOrDefault(subpartitionId, Tuple2.of(null,
-1));
+ ReadableByteChannel channel = fileChannelAndSegmentId.f0;
+
+ // Create the channel if there is a new segment file for a
subpartition.
+ if (channel == null || fileChannelAndSegmentId.f1 != segmentId) {
+ if (channel != null) {
+ channel.close();
+ }
+ channel = openNewChannel(partitionId, subpartitionId, segmentId);
+ if (channel == null) {
+ // return null if the segment file doesn't exist.
+ return null;
+ }
+ subpartitionInfo.put(subpartitionId, Tuple2.of(channel,
segmentId));
+ }
+
+ // Try to read a buffer from the channel.
+ reusedHeaderBuffer.clear();
+ int bufferHeaderResult = channel.read(reusedHeaderBuffer);
+ if (bufferHeaderResult == -1) {
+ channel.close();
+ openedChannelAndSegmentIds.get(partitionId).remove(subpartitionId);
+ return new NetworkBuffer(memorySegment, recycler,
Buffer.DataType.END_OF_SEGMENT);
+ }
+ reusedHeaderBuffer.rewind();
Review Comment:
If `reusedHeaderBuffer.flip()` also applies, I would recommend it more.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileReader.java:
##########
@@ -42,8 +82,44 @@ public Buffer readBuffer(
MemorySegment memorySegment,
BufferRecycler recycler)
throws IOException {
- // TODO, implement the SegmentPartitionFileReader
- return null;
+
+ // Get the channel of the segment file for a subpartition.
+ Map<TieredStorageSubpartitionId, Tuple2<ReadableByteChannel, Integer>>
subpartitionInfo =
+ openedChannelAndSegmentIds.computeIfAbsent(partitionId, ignore
-> new HashMap<>());
+ Tuple2<ReadableByteChannel, Integer> fileChannelAndSegmentId =
+ subpartitionInfo.getOrDefault(subpartitionId, Tuple2.of(null,
-1));
+ ReadableByteChannel channel = fileChannelAndSegmentId.f0;
+
+ // Create the channel if there is a new segment file for a
subpartition.
+ if (channel == null || fileChannelAndSegmentId.f1 != segmentId) {
+ if (channel != null) {
+ channel.close();
+ }
+ channel = openNewChannel(partitionId, subpartitionId, segmentId);
+ if (channel == null) {
+ // return null if the segment file doesn't exist.
+ return null;
+ }
+ subpartitionInfo.put(subpartitionId, Tuple2.of(channel,
segmentId));
+ }
+
+ // Try to read a buffer from the channel.
+ reusedHeaderBuffer.clear();
+ int bufferHeaderResult = channel.read(reusedHeaderBuffer);
+ if (bufferHeaderResult == -1) {
+ channel.close();
+ openedChannelAndSegmentIds.get(partitionId).remove(subpartitionId);
+ return new NetworkBuffer(memorySegment, recycler,
Buffer.DataType.END_OF_SEGMENT);
+ }
+ reusedHeaderBuffer.rewind();
+ BufferHeader header = parseBufferHeader(reusedHeaderBuffer);
+ int dataBufferResult = channel.read(memorySegment.wrap(0,
header.getLength()));
+ if (dataBufferResult == -1) {
+ throw new IOException("Empty data buffer is read.");
Review Comment:
Should we release(close) this channel before throw exception?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.flink.runtime.io.network.partition.hybrid.tiered.tier.remote;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.memory.MemorySegment;
+import org.apache.flink.core.memory.MemorySegmentFactory;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.FreeingBufferRecycler;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.PartitionFileReader;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.AvailabilityNotifier;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.tier.TierConsumerAgent;
+import org.apache.flink.util.ExceptionUtils;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/** The data client is used to fetch data from remote tier. */
+public class RemoteTierConsumerAgent implements TierConsumerAgent {
+
+ private final RemoteStorageScanner remoteStorageScanner;
+
+ private final PartitionFileReader partitionFileReader;
+
+ /**
+ * The current reading buffer indexes and segment ids stored in map.
+ *
+ * <p>The key is partition id and subpartition id. The value is buffer
index and segment id.
+ */
+ private final Map<
+ TieredStoragePartitionId,
+ Map<TieredStorageSubpartitionId, Tuple2<Integer, Integer>>>
+ currentBufferIndexAndSegmentIds;
+
+ private final int bufferSizeBytes;
+
+ public RemoteTierConsumerAgent(
+ RemoteStorageScanner remoteStorageScanner,
+ PartitionFileReader partitionFileReader,
+ int bufferSizeBytes) {
+ this.remoteStorageScanner = remoteStorageScanner;
+ this.currentBufferIndexAndSegmentIds = new HashMap<>();
+ this.partitionFileReader = partitionFileReader;
+ this.bufferSizeBytes = bufferSizeBytes;
+ }
+
+ @Override
+ public void start() {
+ remoteStorageScanner.start();
+ }
+
+ @Override
+ public Optional<Buffer> getNextBuffer(
+ TieredStoragePartitionId partitionId,
+ TieredStorageSubpartitionId subpartitionId,
+ int segmentId) {
+ // Get current segment id and buffer index.
+ Tuple2<Integer, Integer> bufferIndexAndSegmentId =
+ currentBufferIndexAndSegmentIds
+ .computeIfAbsent(partitionId, ignore -> new
HashMap<>())
+ .getOrDefault(subpartitionId, Tuple2.of(0, -1));
+ int currentBufferIndex = bufferIndexAndSegmentId.f0;
+ int currentSegmentId = bufferIndexAndSegmentId.f1;
+ if (segmentId != currentSegmentId) {
+ remoteStorageScanner.watchSegment(partitionId, subpartitionId,
segmentId);
+ }
+
+ // Read buffer from the partition file in remote storage.
+ MemorySegment memorySegment =
MemorySegmentFactory.allocateUnpooledSegment(bufferSizeBytes);
Review Comment:
We should also free this segment if something unexpected happens. I know
that the current implementation will not cause problems as this `byte[]` will
be released during gc even though we manually free it. But this is not a good
pattern.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScannerTest.java:
##########
@@ -0,0 +1,163 @@
+/*
+ * 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.flink.runtime.io.network.partition.hybrid.tiered.tier.remote;
+
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageIdMappingUtils;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+
+import static
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentPath;
+import static
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.writeSegmentFinishFile;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
+
+/** Tests for {@link RemoteStorageScanner}. */
+class RemoteStorageScannerTest {
+
+ private static final TieredStoragePartitionId DEFAULT_PARTITION_ID =
+ TieredStorageIdMappingUtils.convertId(new ResultPartitionID());
+
+ private static final TieredStorageSubpartitionId DEFAULT_SUBPARTITION_ID =
+ new TieredStorageSubpartitionId(0);
+
+ @TempDir private java.nio.file.Path tempFolder;
Review Comment:
```suggestion
@TempDir private File tempFolder;
```
Since we will definitely convert it into a `File`, why not use it directly?
Other tests are the same.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScanner.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.flink.runtime.io.network.partition.hybrid.tiered.tier.remote;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.AvailabilityNotifier;
+import org.apache.flink.util.ExceptionUtils;
+
+import
org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentFinishDirPath;
+import static
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentPath;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * The {@link RemoteStorageScanner} is introduced to notify asynchronously for
file reading on
+ * remote storage. Asynchronous notifications will prevent {@link
RemoteTierConsumerAgent} from
+ * repeatedly attempting to read remote files and reduce CPU consumption.
+ *
+ * <p>It will be invoked by {@link RemoteTierConsumerAgent} to watch the
required segments and scan
+ * the existence status of the segments. If the segment file is found, it will
notify the
+ * availability of segment file.
+ */
+public class RemoteStorageScanner implements Runnable {
+
+ /** The initial scan interval is 100ms. */
+ private static final int INITIAL_SCAN_INTERVAL_MS = 100;
+
+ /** The max scan interval is 10000ms. */
+ private static final int MAX_SCAN_INTERVAL_MS = 10_000;
+
+ /** Executor to scan the existence status of segment files on remote
storage. */
+ private final ScheduledExecutorService scannerExecutor =
+ Executors.newSingleThreadScheduledExecutor(
+ new ThreadFactoryBuilder()
+ .setNameFormat("remote storage file scanner")
+ .build());
+
+ /** The key is partition id and subpartition id, the value is required
segment id. */
+ private final Map<Tuple2<TieredStoragePartitionId,
TieredStorageSubpartitionId>, Integer>
+ requiredSegmentIds;
+
+ /**
+ * The key is partition id and subpartition id, the value is max id of
written segment files in
+ * the subpartition.
+ */
+ private final Map<Tuple2<TieredStoragePartitionId,
TieredStorageSubpartitionId>, Integer>
+ scannedMaxSegmentIds;
+
+ private final String baseRemoteStoragePath;
+
+ private final ScanStrategy scanStrategy;
+
+ private final FileSystem remoteFileSystem;
+
+ private AvailabilityNotifier notifier;
Review Comment:
This should be marked as `Nullable`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/file/SegmentPartitionFileReader.java:
##########
@@ -42,8 +82,44 @@ public Buffer readBuffer(
MemorySegment memorySegment,
BufferRecycler recycler)
throws IOException {
- // TODO, implement the SegmentPartitionFileReader
- return null;
+
+ // Get the channel of the segment file for a subpartition.
+ Map<TieredStorageSubpartitionId, Tuple2<ReadableByteChannel, Integer>>
subpartitionInfo =
+ openedChannelAndSegmentIds.computeIfAbsent(partitionId, ignore
-> new HashMap<>());
+ Tuple2<ReadableByteChannel, Integer> fileChannelAndSegmentId =
+ subpartitionInfo.getOrDefault(subpartitionId, Tuple2.of(null,
-1));
+ ReadableByteChannel channel = fileChannelAndSegmentId.f0;
+
+ // Create the channel if there is a new segment file for a
subpartition.
+ if (channel == null || fileChannelAndSegmentId.f1 != segmentId) {
+ if (channel != null) {
+ channel.close();
+ }
+ channel = openNewChannel(partitionId, subpartitionId, segmentId);
+ if (channel == null) {
+ // return null if the segment file doesn't exist.
+ return null;
+ }
+ subpartitionInfo.put(subpartitionId, Tuple2.of(channel,
segmentId));
+ }
+
+ // Try to read a buffer from the channel.
+ reusedHeaderBuffer.clear();
+ int bufferHeaderResult = channel.read(reusedHeaderBuffer);
+ if (bufferHeaderResult == -1) {
+ channel.close();
+ openedChannelAndSegmentIds.get(partitionId).remove(subpartitionId);
+ return new NetworkBuffer(memorySegment, recycler,
Buffer.DataType.END_OF_SEGMENT);
+ }
+ reusedHeaderBuffer.rewind();
+ BufferHeader header = parseBufferHeader(reusedHeaderBuffer);
+ int dataBufferResult = channel.read(memorySegment.wrap(0,
header.getLength()));
+ if (dataBufferResult == -1) {
Review Comment:
Maybe we should check the `dataBufferResult` is equals to
`header.getLength()` instead of empty.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteStorageScanner.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.flink.runtime.io.network.partition.hybrid.tiered.tier.remote;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStoragePartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.common.TieredStorageSubpartitionId;
+import
org.apache.flink.runtime.io.network.partition.hybrid.tiered.storage.AvailabilityNotifier;
+import org.apache.flink.util.ExceptionUtils;
+
+import
org.apache.flink.shaded.guava31.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentFinishDirPath;
+import static
org.apache.flink.runtime.io.network.partition.hybrid.tiered.file.SegmentPartitionFile.getSegmentPath;
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * The {@link RemoteStorageScanner} is introduced to notify asynchronously for
file reading on
+ * remote storage. Asynchronous notifications will prevent {@link
RemoteTierConsumerAgent} from
+ * repeatedly attempting to read remote files and reduce CPU consumption.
+ *
+ * <p>It will be invoked by {@link RemoteTierConsumerAgent} to watch the
required segments and scan
+ * the existence status of the segments. If the segment file is found, it will
notify the
+ * availability of segment file.
+ */
+public class RemoteStorageScanner implements Runnable {
+
+ /** The initial scan interval is 100ms. */
+ private static final int INITIAL_SCAN_INTERVAL_MS = 100;
+
+ /** The max scan interval is 10000ms. */
+ private static final int MAX_SCAN_INTERVAL_MS = 10_000;
+
+ /** Executor to scan the existence status of segment files on remote
storage. */
+ private final ScheduledExecutorService scannerExecutor =
+ Executors.newSingleThreadScheduledExecutor(
Review Comment:
What are we going to do with UncaughtException?
--
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]
