You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by pe...@apache.org on 2026/10/24 15:04:09 UTC
(ozone) branch master updated: HDDS-16751. Fix possible NPE in KeyInputStream#getBlockLocationInfo during block location refresh (#11435)
This is an automated email from the ASF dual-hosted git repository.
peterxcli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 7476f23e5d0 HDDS-16751. Fix possible NPE in KeyInputStream#getBlockLocationInfo during block location refresh (#11435)
7476f23e5d0 is described below
commit 7476f23e5d0557ff4451b952a08f760ccd434599
Author: Neo Chien <67...@users.noreply.github.com>
AuthorDate: Sat Oct 24 23:04:03 2026 +0800
HDDS-16751. Fix possible NPE in KeyInputStream#getBlockLocationInfo during block location refresh (#11435)
---
.../hadoop/ozone/client/io/KeyInputStream.java | 18 ++--
.../apache/hadoop/ozone/client/rpc/RpcClient.java | 18 ++--
.../hadoop/ozone/client/io/TestKeyInputStream.java | 98 ++++++++++++++++++++++
.../org/apache/hadoop/ozone/TestBlockTokens.java | 12 +--
4 files changed, 118 insertions(+), 28 deletions(-)
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyInputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyInputStream.java
index c11514f8fd4..3fce123111c 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyInputStream.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyInputStream.java
@@ -36,6 +36,7 @@
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.apache.ratis.util.function.CheckedFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,7 +57,7 @@ private static List<BlockExtendedInputStream> createStreams(
OmKeyInfo keyInfo,
List<OmKeyLocationInfo> blockInfos,
XceiverClientFactory xceiverClientFactory,
- Function<OmKeyInfo, OmKeyInfo> retryFunction,
+ CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunction,
BlockInputStreamFactory blockStreamFactory,
OzoneClientConfig config) throws IOException {
boolean isHsyncFile = keyInfo.getMetadata().containsKey(OzoneConsts.HSYNC_CLIENT_ID);
@@ -77,9 +78,12 @@ private static List<BlockExtendedInputStream> createStreams(
Function<BlockID, BlockLocationInfo> retry;
if (retryFunction != null) {
retry = keyBlockID -> {
- OmKeyInfo newKeyInfo = retryFunction.apply(keyInfo);
- return getBlockLocationInfo(newKeyInfo,
- omKeyLocationInfo.getBlockID());
+ try {
+ return getBlockLocationInfo(retryFunction.apply(keyInfo), omKeyLocationInfo.getBlockID());
+ } catch (IOException e) {
+ LOG.error("Unable to lookup key {} on retry.", keyInfo.getKeyName(), e);
+ return null;
+ }
};
} else {
retry = null;
@@ -120,7 +124,7 @@ private static BlockLocationInfo getBlockLocationInfo(OmKeyInfo newKeyInfo,
private static LengthInputStream getFromOmKeyInfo(
OmKeyInfo keyInfo,
XceiverClientFactory xceiverClientFactory,
- Function<OmKeyInfo, OmKeyInfo> retryFunction,
+ CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunction,
BlockInputStreamFactory blockStreamFactory,
List<OmKeyLocationInfo> locationInfos,
OzoneClientConfig config) throws IOException {
@@ -137,7 +141,7 @@ private static LengthInputStream getFromOmKeyInfo(
*/
public static LengthInputStream getFromOmKeyInfo(OmKeyInfo keyInfo,
XceiverClientFactory xceiverClientFactory,
- Function<OmKeyInfo, OmKeyInfo> retryFunction,
+ CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunction,
BlockInputStreamFactory blockStreamFactory,
OzoneClientConfig config) throws IOException {
@@ -150,7 +154,7 @@ public static LengthInputStream getFromOmKeyInfo(OmKeyInfo keyInfo,
public static List<LengthInputStream> getStreamsFromKeyInfo(OmKeyInfo keyInfo,
XceiverClientFactory xceiverClientFactory,
- Function<OmKeyInfo, OmKeyInfo> retryFunction,
+ CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunction,
BlockInputStreamFactory blockStreamFactory,
OzoneClientConfig config) throws IOException {
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
index fdb1cb16261..88014d62e61 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
@@ -58,7 +58,6 @@
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
import java.util.stream.Collectors;
import javax.crypto.Cipher;
import javax.crypto.CipherInputStream;
@@ -194,6 +193,7 @@
import org.apache.hadoop.util.Time;
import org.apache.ratis.protocol.ClientId;
import org.apache.ratis.util.MemoizedSupplier;
+import org.apache.ratis.util.function.CheckedFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1810,7 +1810,7 @@ public OzoneInputStream getKey(
.build();
dnKeyInfo.setKeyLocationVersions(keyLocationInfoGroups);
- blocks.put(dn, createInputStream(dnKeyInfo, Function.identity()));
+ blocks.put(dn, createInputStream(dnKeyInfo, k -> k));
}
result.put(locationInfo, blocks);
@@ -2582,15 +2582,9 @@ private OzoneInputStream getInputStreamWithRetryFunction(
*/
private OzoneInputStream getInputStreamWithRetryFunction(
OmKeyInfo keyInfo, boolean requireFileSystemLayout) throws IOException {
- return createInputStream(keyInfo, omKeyInfo -> {
- try {
- return getKeyInfo(omKeyInfo.getVolumeName(), omKeyInfo.getBucketName(),
- omKeyInfo.getKeyName(), true, requireFileSystemLayout);
- } catch (IOException e) {
- LOG.error("Unable to lookup key {} on retry.", keyInfo.getKeyName(), e);
- return null;
- }
- });
+ return createInputStream(keyInfo, omKeyInfo -> getKeyInfo(omKeyInfo.getVolumeName(),
+ omKeyInfo.getBucketName(), omKeyInfo.getKeyName(), true,
+ requireFileSystemLayout));
}
@Override
@@ -2792,7 +2786,7 @@ static GDPRSymmetricKey getGDPRSymmetricKey(Map<String, String> metadata,
}
private OzoneInputStream createInputStream(
- OmKeyInfo keyInfo, Function<OmKeyInfo, OmKeyInfo> retryFunction)
+ OmKeyInfo keyInfo, CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunction)
throws IOException {
// When Key is not MPU or when Key is MPU and encryption is not enabled
// Need to revisit for GDP.
diff --git a/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyInputStream.java b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyInputStream.java
new file mode 100644
index 00000000000..95ea15b26b8
--- /dev/null
+++ b/hadoop-ozone/client/src/test/java/org/apache/hadoop/ozone/client/io/TestKeyInputStream.java
@@ -0,0 +1,98 @@
+/*
+ * 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.hadoop.ozone.client.io;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
+import org.apache.ratis.util.function.CheckedFunction;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests the block location refresh function that {@link KeyInputStream} gives to each block stream.
+ */
+public class TestKeyInputStream {
+
+ private static final BlockID BLOCK_ID = new BlockID(1, 1);
+
+ @Test
+ public void refreshReturnsNullWhenKeyLookupFails() throws IOException {
+ Function<BlockID, BlockLocationInfo> refresh = getRefreshFunction(keyInfo -> {
+ throw new IOException("lookup failed");
+ });
+
+ assertNull(refresh.apply(BLOCK_ID));
+ }
+
+ @Test
+ public void refreshReturnsLocationFromRefreshedKey() throws IOException {
+ OmKeyLocationInfo refreshed = newLocation(BLOCK_ID);
+ Function<BlockID, BlockLocationInfo> refresh = getRefreshFunction(keyInfo -> newKeyInfo(refreshed));
+
+ assertSame(refreshed, refresh.apply(BLOCK_ID));
+ }
+
+ @Test
+ public void refreshReturnsNullWhenBlockIsNotInRefreshedKey() throws IOException {
+ Function<BlockID, BlockLocationInfo> refresh =
+ getRefreshFunction(keyInfo -> newKeyInfo(newLocation(new BlockID(2, 2))));
+
+ assertNull(refresh.apply(BLOCK_ID));
+ }
+
+ private static Function<BlockID, BlockLocationInfo> getRefreshFunction(
+ CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunction) throws IOException {
+ AtomicReference<Function<BlockID, BlockLocationInfo>> refresh = new AtomicReference<>();
+ BlockInputStreamFactory factory = (repConfig, blockInfo, pipeline, token, xceiverFactory, refreshFunction,
+ config) -> {
+ refresh.set(refreshFunction);
+ return mock(BlockExtendedInputStream.class);
+ };
+ KeyInputStream.getFromOmKeyInfo(newKeyInfo(newLocation(BLOCK_ID)), null, retryFunction, factory,
+ new OzoneClientConfig()).close();
+ return refresh.get();
+ }
+
+ private static OmKeyLocationInfo newLocation(BlockID blockID) {
+ return new OmKeyLocationInfo.Builder().setBlockID(blockID).build();
+ }
+
+ private static OmKeyInfo newKeyInfo(OmKeyLocationInfo location) {
+ return new OmKeyInfo.Builder()
+ .setVolumeName("vol")
+ .setBucketName("bucket")
+ .setKeyName("key")
+ .setReplicationConfig(RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .addOmKeyLocationInfoGroup(new OmKeyLocationInfoGroup(0, Collections.singletonList(location)))
+ .build();
+ }
+}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestBlockTokens.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestBlockTokens.java
index 25def077299..d85cb056edd 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestBlockTokens.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestBlockTokens.java
@@ -56,7 +56,6 @@
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.TimeoutException;
-import java.util.function.Function;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.hadoop.hdds.annotation.InterfaceAudience;
@@ -86,6 +85,7 @@
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.token.Token;
import org.apache.ratis.util.ExitUtils;
+import org.apache.ratis.util.function.CheckedFunction;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -205,13 +205,7 @@ public void blockTokenOnExpiredSecretKeyRetrySuccessful() throws Exception {
EXPIRY_DURATION_IN_MS);
assertTrue(secretKey.isExpired());
// verify that the read is denied because of the expired secret key.
- readData(keyInfo, k -> {
- try {
- return getTestKeyInfo();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- });
+ readData(keyInfo, k -> getTestKeyInfo());
}
@Test
@@ -283,7 +277,7 @@ private void readDataWithoutRetry(OmKeyInfo keyInfo) throws IOException {
}
private void readData(OmKeyInfo keyInfo,
- Function<OmKeyInfo, OmKeyInfo> retryFunc) throws IOException {
+ CheckedFunction<OmKeyInfo, OmKeyInfo, IOException> retryFunc) throws IOException {
XceiverClientFactory xceiverClientManager =
((RpcClient) client.getProxy()).getXceiverClientManager();
OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org