Support OpenSSL Provider with default Netty allocator (#5460)

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

CHANGELOG.md

@@ -32,6 +32,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Fixed
 - Fix 1.x compatibility bug with stored Tasks ([#5412](https://github.com/opensearch-project/OpenSearch/pull/5412))
 - Fix case sensitivity for wildcard queries ([#5462](https://github.com/opensearch-project/OpenSearch/pull/5462))
+- Support OpenSSL Provider with default Netty allocator ([#5460](https://github.com/opensearch-project/OpenSearch/pull/5460))
 ### Security
 
 [Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.4...2.x

modules/transport-netty4/src/main/java/org/opensearch/transport/Netty4NioServerSocketChannel.java

@@ -0,0 +1,62 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.transport;
+
+import io.netty.channel.socket.InternetProtocolFamily;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.util.internal.SocketUtils;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
+
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.spi.SelectorProvider;
+import java.util.List;
+
+public class Netty4NioServerSocketChannel extends NioServerSocketChannel {
+ private static final InternalLogger logger = InternalLoggerFactory.getInstance(Netty4NioServerSocketChannel.class);
+
+ public Netty4NioServerSocketChannel() {
+ super();
+ }
+
+ public Netty4NioServerSocketChannel(SelectorProvider provider) {
+ super(provider);
+ }
+
+ public Netty4NioServerSocketChannel(SelectorProvider provider, InternetProtocolFamily family) {
+ super(provider, family);
+ }
+
+ public Netty4NioServerSocketChannel(ServerSocketChannel channel) {
+ super(channel);
+ }
+
+ @Override
+ protected int doReadMessages(List<Object> buf) throws Exception {
+ SocketChannel ch = SocketUtils.accept(javaChannel());
+
+ try {
+ if (ch != null) {
+ buf.add(new Netty4NioSocketChannel(this, ch));
+ return 1;
+ }
+ } catch (Throwable t) {
+ logger.warn("Failed to create a new channel from an accepted socket.", t);
+
+ try {
+ ch.close();
+ } catch (Throwable t2) {
+ logger.warn("Failed to close a socket.", t2);
+ }
+ }
+
+ return 0;
+ }
+}

modules/transport-netty4/src/main/java/org/opensearch/transport/NettyAllocator.java

@@ -39,7 +39,6 @@
 import io.netty.buffer.UnpooledByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ServerChannel;
-import io.netty.channel.socket.nio.NioServerSocketChannel;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.common.Booleans;
@@ -181,7 +180,7 @@ public static Class<? extends ServerChannel> getServerChannelType() {
  if (ALLOCATOR instanceof NoDirectBuffers) {
  return CopyBytesServerSocketChannel.class;
  } else {
- return NioServerSocketChannel.class;
+ return Netty4NioServerSocketChannel.class;
  }
  }
 

server/src/main/java/org/opensearch/common/bytes/BytesReference.java

@@ -122,8 +122,13 @@ static BytesReference fromByteBuffers(ByteBuffer[] buffers) {
  * Returns BytesReference composed of the provided ByteBuffer.
  */
  static BytesReference fromByteBuffer(ByteBuffer buffer) {
- assert buffer.hasArray();
- return new BytesArray(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining());
+ if (buffer.hasArray()) {
+ return new BytesArray(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining());
+ } else {
+ final byte[] array = new byte[buffer.remaining()];
+ buffer.asReadOnlyBuffer().get(array, 0, buffer.remaining());
+ return new BytesArray(array);
+ }
  }
 
  /**

server/src/test/java/org/opensearch/common/bytes/ByteBuffersBytesReferenceTests.java

@@ -0,0 +1,79 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.common.bytes;
+
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.hamcrest.Matchers;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.function.Function;
+
+public class ByteBuffersBytesReferenceTests extends AbstractBytesReferenceTestCase {
+ @ParametersFactory
+ public static Collection<Object[]> allocator() {
+ return Arrays.asList(
+ new Object[] { (Function<Integer, ByteBuffer>) ByteBuffer::allocateDirect },
+ new Object[] { (Function<Integer, ByteBuffer>) ByteBuffer::allocate }
+ );
+ }
+
+ private final Function<Integer, ByteBuffer> allocator;
+
+ public ByteBuffersBytesReferenceTests(Function<Integer, ByteBuffer> allocator) {
+ this.allocator = allocator;
+ }
+
+ @Override
+ protected BytesReference newBytesReference(int length) throws IOException {
+ return newBytesReference(length, randomInt(length));
+ }
+
+ @Override
+ protected BytesReference newBytesReferenceWithOffsetOfZero(int length) throws IOException {
+ return newBytesReference(length, 0);
+ }
+
+ private BytesReference newBytesReference(int length, int offset) throws IOException {
+ // we know bytes stream output always creates a paged bytes reference, we use it to create randomized content
+ final ByteBuffer buffer = allocator.apply(length + offset);
+ for (int i = 0; i < length + offset; i++) {
+ buffer.put((byte) random().nextInt(1 << 8));
+ }
+ assertEquals(length + offset, buffer.limit());
+ buffer.flip().position(offset);
+
+ BytesReference ref = BytesReference.fromByteBuffer(buffer);
+ assertEquals(length, ref.length());
+ assertTrue(ref instanceof BytesArray);
+ assertThat(ref.length(), Matchers.equalTo(length));
+ return ref;
+ }
+
+ public void testArray() throws IOException {
+ int[] sizes = { 0, randomInt(PAGE_SIZE), PAGE_SIZE, randomIntBetween(2, PAGE_SIZE * randomIntBetween(2, 5)) };
+
+ for (int i = 0; i < sizes.length; i++) {
+ BytesArray pbr = (BytesArray) newBytesReference(sizes[i]);
+ byte[] array = pbr.array();
+ assertNotNull(array);
+ assertEquals(sizes[i], array.length - pbr.offset());
+ assertSame(array, pbr.array());
+ }
+ }
+
+ public void testArrayOffset() throws IOException {
+ int length = randomInt(PAGE_SIZE * randomIntBetween(2, 5));
+ BytesArray pbr = (BytesArray) newBytesReferenceWithOffsetOfZero(length);
+ assertEquals(0, pbr.offset());
+ }
+}