This is an automated email from the ASF dual-hosted git repository.
adoroszlai 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 dae6f30a79 HDDS-8095. Unbuffer not supported in TDE buckets (#4361)
dae6f30a79 is described below
commit dae6f30a79fa1edfbe6a6ce127a9329025888bbf
Author: Wei-Chiu Chuang <[email protected]>
AuthorDate: Thu Mar 9 08:17:24 2023 -0800
HDDS-8095. Unbuffer not supported in TDE buckets (#4361)
---
.../hdds/scm/storage/ExtendedInputStream.java | 15 ++-
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 34 ++++++
.../org/apache/hadoop/fs/StreamCapabilities.java | 132 +++++++++++++++++++++
.../java/org/apache/hadoop/fs/package-info.java | 29 +++++
4 files changed, 209 insertions(+), 1 deletion(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
index 7f693c70f9..3e78abbf48 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java
@@ -21,6 +21,8 @@ import org.apache.commons.lang3.NotImplementedException;
import org.apache.hadoop.fs.ByteBufferReadable;
import org.apache.hadoop.fs.CanUnbuffer;
import org.apache.hadoop.fs.Seekable;
+import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.util.StringUtils;
import java.io.IOException;
import java.io.InputStream;
@@ -31,7 +33,7 @@ import java.nio.ByteBuffer;
* various Ozone InputStream classes.
*/
public abstract class ExtendedInputStream extends InputStream
- implements Seekable, CanUnbuffer, ByteBufferReadable {
+ implements Seekable, CanUnbuffer, ByteBufferReadable, StreamCapabilities {
protected static final int EOF = -1;
@@ -88,4 +90,15 @@ public abstract class ExtendedInputStream extends InputStream
public synchronized boolean seekToNewSource(long l) throws IOException {
return false;
}
+
+ @Override
+ public boolean hasCapability(String capability) {
+ switch (StringUtils.toLowerCase(capability)) {
+ case StreamCapabilities.READBYTEBUFFER:
+ case StreamCapabilities.UNBUFFER:
+ return true;
+ default:
+ return false;
+ }
+ }
}
diff --git
a/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
b/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
index 90d3948d57..3a1d6daace 100644
---
a/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
+++
b/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
@@ -19,14 +19,22 @@ package org.apache.hadoop.fs.ozone;
import com.google.common.collect.ImmutableList;
import org.apache.commons.lang3.RandomUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.CipherSuite;
+import org.apache.hadoop.crypto.CryptoCodec;
+import org.apache.hadoop.crypto.CryptoInputStream;
+import org.apache.hadoop.crypto.Decryptor;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.ozone.client.io.KeyInputStream;
+
import org.junit.Test;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.security.GeneralSecurityException;
import java.util.Arrays;
import java.util.List;
import java.util.function.IntFunction;
@@ -34,6 +42,11 @@ import java.util.function.IntFunction;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/**
* Tests for {@link OzoneFSInputStream}.
@@ -132,6 +145,27 @@ public class TestOzoneFSInputStream {
}
}
+ @Test
+ public void testCryptoStreamUnbuffer()
+ throws IOException, GeneralSecurityException {
+ KeyInputStream keyInputStream = mock(KeyInputStream.class);
+ when(keyInputStream.hasCapability(anyString())).thenReturn(true);
+
+ CryptoCodec codec = mock(CryptoCodec.class);
+ when(codec.getCipherSuite()).thenReturn(CipherSuite.AES_CTR_NOPADDING);
+ when(codec.getConf()).thenReturn(new Configuration());
+ Decryptor decryptor = mock(Decryptor.class);
+ when(codec.createDecryptor()).thenReturn(decryptor);
+ CryptoInputStream cis = new CryptoInputStream(keyInputStream, codec,
+ new byte[0], new byte[0]);
+ try {
+ cis.unbuffer();
+ verify(keyInputStream, times(1)).unbuffer();
+ } finally {
+ cis.close();
+ }
+ }
+
private static OzoneFSInputStream createTestSubject(InputStream input) {
return new OzoneFSInputStream(input,
new FileSystem.Statistics("test"));
diff --git
a/hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
b/hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
new file mode 100644
index 0000000000..810cd39533
--- /dev/null
+++
b/hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
@@ -0,0 +1,132 @@
+/**
+ * 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.fs;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Interface to query streams for supported capabilities.
+ *
+ * This is copied from Hadoop3 code to make Ozone compatible with Hadoop2.7
+ * runtime dependency. Hadoop 2.7 does not support StreamCapabilities API.
+ *
+ * TODO: remove this interface once we drop Hadoop 2.7 support. (HDDS-8113)
+ *
+ * Capability strings must be in lower case.
+ *
+ * Constant strings are chosen over enums in order to allow other file systems
+ * to define their own capabilities.
+ */
[email protected]
[email protected]
+public interface StreamCapabilities {
+ /**
+ * Stream hflush capability implemented by {@link Syncable#hflush()}.
+ *
+ * Use the {@link #HSYNC} probe to check for the support of Syncable;
+ * it's that presence of {@code hsync()} which matters.
+ */
+ @Deprecated
+ String HFLUSH = "hflush";
+
+ /**
+ * Stream hsync capability implemented by {@link Syncable#hsync()}.
+ */
+ String HSYNC = "hsync";
+
+ /**
+ * Stream setReadahead capability implemented by
+ * {@link CanSetReadahead#setReadahead(Long)}.
+ */
+ String READAHEAD = "in:readahead";
+
+ /**
+ * Stream setDropBehind capability implemented by
+ * {@link CanSetDropBehind#setDropBehind(Boolean)}.
+ */
+ String DROPBEHIND = "dropbehind";
+
+ /**
+ * Stream unbuffer capability implemented by {@link CanUnbuffer#unbuffer()}.
+ */
+ String UNBUFFER = "in:unbuffer";
+
+ /**
+ * Stream read(ByteBuffer) capability implemented by
+ * {@link ByteBufferReadable#read(java.nio.ByteBuffer)}.
+ */
+ String READBYTEBUFFER = "in:readbytebuffer";
+
+ /**
+ * Stream read(long, ByteBuffer) capability implemented by
+ * ByteBufferPositionedReadable.
+ */
+ String PREADBYTEBUFFER = "in:preadbytebuffer";
+
+ /**
+ * IOStatisticsSource API.
+ */
+ String IOSTATISTICS = "iostatistics";
+
+ /**
+ * Support for vectored IO api.
+ * See {@code PositionedReadable#readVectored(List, IntFunction)}.
+ */
+ String VECTOREDIO = "in:readvectored";
+
+ /**
+ * Stream abort() capability implemented by Abortable.
+ */
+ String ABORTABLE_STREAM = "fs.capability.outputstream.abortable";
+
+ /**
+ * Streams that support IOStatistics context and capture thread-level
+ * IOStatistics.
+ */
+ String IOSTATISTICS_CONTEXT = "fs.capability.iocontext.supported";
+
+ /**
+ * Capabilities that a stream can support and be queried for.
+ */
+ @Deprecated
+ enum StreamCapability {
+ HFLUSH(StreamCapabilities.HFLUSH),
+ HSYNC(StreamCapabilities.HSYNC);
+
+ private final String capability;
+
+ StreamCapability(String value) {
+ this.capability = value;
+ }
+
+ public final String getValue() {
+ return capability;
+ }
+ }
+
+ /**
+ * Query the stream for a specific capability.
+ *
+ * @param capability string to query the stream support for.
+ * @return True if the stream supports capability.
+ */
+ boolean hasCapability(String capability);
+}
+
diff --git
a/hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/package-info.java
b/hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/package-info.java
new file mode 100644
index 0000000000..b29b1399da
--- /dev/null
+++
b/hadoop-ozone/ozonefs-hadoop2/src/main/java/org/apache/hadoop/fs/package-info.java
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ *
+ */
+
+/**
+ * Utility classes copied from Hadoop3 to maintain Hadoop2 runtime
+ * compatibility.
+ */
[email protected]
[email protected]
+package org.apache.hadoop.fs;
+
+import org.apache.hadoop.hdds.annotation.InterfaceAudience;
+import org.apache.hadoop.hdds.annotation.InterfaceStability;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
