This is an automated email from the ASF dual-hosted git repository.
guoweijie pushed a commit to branch release-1.18
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.18 by this push:
new 0501d036584 [FLINK-33088][network] Fix NullPointerException in
RemoteTierConsumerAgent for tiered storage
0501d036584 is described below
commit 0501d03658495b86b33ea57913c4baa131287388
Author: Yuxin Tan <[email protected]>
AuthorDate: Thu Sep 14 15:51:41 2023 +0800
[FLINK-33088][network] Fix NullPointerException in RemoteTierConsumerAgent
for tiered storage
---
.../tier/remote/RemoteTierConsumerAgent.java | 5 +-
.../tier/remote/RemoteTierConsumerAgentTest.java | 93 ++++++++++++++++++++++
2 files changed, 95 insertions(+), 3 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java
index 12d046e8325..7723046c256 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgent.java
@@ -36,7 +36,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
/** The data client is used to fetch data from remote tier. */
@@ -107,8 +106,8 @@ public class RemoteTierConsumerAgent implements
TierConsumerAgent {
memorySegment.free();
ExceptionUtils.rethrow(e, "Failed to read buffer from partition
file.");
}
- List<Buffer> readBuffers =
checkNotNull(readBufferResult).getReadBuffers();
- if (!readBuffers.isEmpty()) {
+ if (readBufferResult != null &&
!readBufferResult.getReadBuffers().isEmpty()) {
+ List<Buffer> readBuffers = readBufferResult.getReadBuffers();
checkState(readBuffers.size() == 1);
Buffer buffer = readBuffers.get(0);
currentBufferIndexAndSegmentIds
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgentTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgentTest.java
new file mode 100644
index 00000000000..eeb53f901e2
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/hybrid/tiered/tier/remote/RemoteTierConsumerAgentTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.Path;
+import org.apache.flink.runtime.io.network.buffer.Buffer;
+import org.apache.flink.runtime.io.network.buffer.BufferBuilderTestUtils;
+import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+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.file.TestingPartitionFileReader;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.Collections;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RemoteTierConsumerAgent}. */
+class RemoteTierConsumerAgentTest {
+
+ @TempDir private File tempFolder;
+
+ private String remoteStoragePath;
+
+ @BeforeEach
+ void before() {
+ remoteStoragePath = Path.fromLocalFile(tempFolder).getPath();
+ }
+
+ @Test
+ void testGetEmptyBuffer() {
+ RemoteTierConsumerAgent remoteTierConsumerAgent =
+ new RemoteTierConsumerAgent(
+ new RemoteStorageScanner(remoteStoragePath),
+ new TestingPartitionFileReader.Builder().build(),
+ 1024);
+ assertThat(
+ remoteTierConsumerAgent.getNextBuffer(
+ new TieredStoragePartitionId(new
ResultPartitionID()),
+ new TieredStorageSubpartitionId(0),
+ 0))
+ .isEmpty();
+ }
+
+ @Test
+ void testGetBuffer() {
+ int bufferSize = 10;
+ PartitionFileReader partitionFileReader =
+ new TestingPartitionFileReader.Builder()
+ .setReadBufferSupplier(
+ (bufferIndex, segmentId) ->
+ new
PartitionFileReader.ReadBufferResult(
+ Collections.singletonList(
+
BufferBuilderTestUtils.buildSomeBuffer(
+ bufferSize)),
+ false,
+ null))
+ .build();
+ RemoteTierConsumerAgent remoteTierConsumerAgent =
+ new RemoteTierConsumerAgent(
+ new RemoteStorageScanner(remoteStoragePath),
partitionFileReader, 1024);
+ Optional<Buffer> optionalBuffer =
+ remoteTierConsumerAgent.getNextBuffer(
+ new TieredStoragePartitionId(new ResultPartitionID()),
+ new TieredStorageSubpartitionId(0),
+ 0);
+ assertThat(optionalBuffer)
+ .hasValueSatisfying(
+ buffer ->
assertThat(buffer.readableBytes()).isEqualTo(bufferSize));
+ }
+}
