Skip to content

Commit 53ad56f

Browse files
rishabhmauryarishabhmaurya
committed
Arrow Flight Server bootstrap logic
* new plugin for StreamManager implementation * integration with server module * support for SslContext in Flight server and client * ClientManager for creating a pool of flight clients for data nodes * custom event loop group and thread pool for server and client channel
1 parent c0f7806 commit 53ad56f

File tree

104 files changed

+10101
-13
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

104 files changed

+10101
-13
lines changed

‎buildSrc/src/main/java/org/opensearch/gradle/testclusters/OpenSearchNode.java

+7
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,7 @@ public class OpenSearchNode implements TestClusterConfiguration {
174174
private boolean isWorkingDirConfigured = false;
175175
private String httpPort = "0";
176176
private String transportPort = "0";
177+
private String streamPort = "0";
177178
private Path confPathData;
178179
private String keystorePassword = "";
179180
private boolean preserveDataDir = false;
@@ -1175,6 +1176,8 @@ private void createConfiguration() {
11751176
baseConfig.put("node.portsfile", "true");
11761177
baseConfig.put("http.port", httpPort);
11771178
baseConfig.put("transport.port", transportPort);
1179+
baseConfig.put("node.attr.transport.stream.port", streamPort);
1180+
11781181
// Default the watermarks to absurdly low to prevent the tests from failing on nodes without enough disk space
11791182
baseConfig.put("cluster.routing.allocation.disk.watermark.low", "1b");
11801183
baseConfig.put("cluster.routing.allocation.disk.watermark.high", "1b");
@@ -1447,6 +1450,10 @@ void setTransportPort(String transportPort) {
14471450
this.transportPort = transportPort;
14481451
}
14491452

1453+
void setStreamPort(String streamPort) {
1454+
this.streamPort = streamPort;
1455+
}
1456+
14501457
void setDataPath(Path dataPath) {
14511458
this.confPathData = dataPath;
14521459
}

‎buildSrc/src/main/java/org/opensearch/gradle/testclusters/RunTask.java

+7
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ public class RunTask extends DefaultTestClustersTask {
6161
public static final String CUSTOM_SETTINGS_PREFIX = "tests.opensearch.";
6262
private static final int DEFAULT_HTTP_PORT = 9200;
6363
private static final int DEFAULT_TRANSPORT_PORT = 9300;
64+
private static final int DEFAULT_STREAM_PORT = 9880;
6465
private static final int DEFAULT_DEBUG_PORT = 5005;
6566
public static final String LOCALHOST_ADDRESS_PREFIX = "127.0.0.1:";
6667

@@ -140,6 +141,8 @@ public void beforeStart() {
140141
int debugPort = DEFAULT_DEBUG_PORT;
141142
int httpPort = DEFAULT_HTTP_PORT;
142143
int transportPort = DEFAULT_TRANSPORT_PORT;
144+
int streamPort = DEFAULT_STREAM_PORT;
145+
143146
Map<String, String> additionalSettings = System.getProperties()
144147
.entrySet()
145148
.stream()
@@ -164,15 +167,19 @@ public void beforeStart() {
164167
firstNode.setHttpPort(String.valueOf(httpPort));
165168
httpPort++;
166169
firstNode.setTransportPort(String.valueOf(transportPort));
170+
firstNode.setStreamPort(String.valueOf(streamPort));
167171
transportPort++;
172+
streamPort++;
168173
firstNode.setting("discovery.seed_hosts", LOCALHOST_ADDRESS_PREFIX + DEFAULT_TRANSPORT_PORT);
169174
cluster.setPreserveDataDir(preserveData);
170175
for (OpenSearchNode node : cluster.getNodes()) {
171176
if (node != firstNode) {
172177
node.setHttpPort(String.valueOf(httpPort));
173178
httpPort++;
174179
node.setTransportPort(String.valueOf(transportPort));
180+
node.setStreamPort(String.valueOf(streamPort));
175181
transportPort++;
182+
streamPort++;
176183
node.setting("discovery.seed_hosts", LOCALHOST_ADDRESS_PREFIX + DEFAULT_TRANSPORT_PORT);
177184
}
178185
additionalSettings.forEach(node::setting);

‎distribution/archives/build.gradle

+2-2
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@
2828
* under the License.
2929
*/
3030

31-
import org.opensearch.gradle.JavaPackageType
31+
import org.opensearch.gradle.JavaPackageType
3232

3333
apply plugin: 'opensearch.internal-distribution-archive-setup'
3434

@@ -190,7 +190,7 @@ distribution_archives {
190190
}
191191
}
192192

193-
193+
194194
linuxPpc64leTar {
195195
archiveClassifier = 'linux-ppc64le'
196196
content {

‎distribution/src/config/jvm.options

+1
Original file line numberDiff line numberDiff line change
@@ -89,3 +89,4 @@ ${error.file}
8989
# See please https://bugs.openjdk.org/browse/JDK-8341127 (openjdk/jdk#21283)
9090
23:-XX:CompileCommand=dontinline,java/lang/invoke/MethodHandle.setAsTypeCache
9191
23:-XX:CompileCommand=dontinline,java/lang/invoke/MethodHandle.asTypeUncached
92+
--add-opens=java.base/java.nio=org.apache.arrow.memory.core,ALL-UNNAMED

‎gradle/libs.versions.toml

+1-1
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ opentelemetry = "1.41.0"
8282
opentelemetrysemconv = "1.27.0-alpha"
8383

8484
# arrow dependencies
85-
arrow = "17.0.0"
85+
arrow = "18.1.0"
8686
flatbuffers = "2.0.0"
8787

8888
[libraries]

‎modules/arrow-flight-rpc/build.gradle

+259
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,259 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*
8+
* Modifications Copyright OpenSearch Contributors. See
9+
* GitHub history for details.
10+
*/
11+
12+
apply plugin: 'opensearch.publish'
13+
apply plugin: 'opensearch.internal-cluster-test'
14+
15+
opensearchplugin {
16+
description 'Arrow flight based Stream implementation'
17+
classname 'org.opensearch.arrow.flight.FlightStreamPlugin'
18+
}
19+
20+
dependencies {
21+
implementation project(':libs:opensearch-arrow-spi')
22+
23+
implementation "io.netty:netty-buffer:${versions.netty}"
24+
implementation "io.netty:netty-codec:${versions.netty}"
25+
implementation "io.netty:netty-codec-http:${versions.netty}"
26+
implementation "io.netty:netty-codec-http2:${versions.netty}"
27+
implementation "io.netty:netty-common:${versions.netty}"
28+
implementation "io.netty:netty-handler:${versions.netty}"
29+
implementation "io.netty:netty-resolver:${versions.netty}"
30+
implementation "io.netty:netty-transport:${versions.netty}"
31+
implementation "io.netty:netty-transport-native-unix-common:${versions.netty}"
32+
implementation "io.netty:netty-transport-classes-epoll:${versions.netty}"
33+
34+
implementation 'org.checkerframework:checker-qual:3.44.0'
35+
implementation "org.apache.arrow:arrow-memory-core:${versions.arrow}"
36+
implementation "org.apache.arrow:arrow-memory-netty-buffer-patch:${versions.arrow}"
37+
implementation "org.apache.arrow:arrow-memory-netty:${versions.arrow}"
38+
39+
runtimeOnly group: 'com.google.code.findbugs', name: 'jsr305', version: '3.0.2'
40+
compileOnly 'org.immutables:value:2.10.1'
41+
annotationProcessor 'org.immutables:value:2.10.1'
42+
implementation "org.apache.arrow:arrow-flight:${versions.arrow}"
43+
implementation "org.apache.arrow:flight-core:${versions.arrow}"
44+
implementation "io.grpc:grpc-api:${versions.grpc}"
45+
implementation "io.grpc:grpc-netty:${versions.grpc}"
46+
runtimeOnly "io.grpc:grpc-core:${versions.grpc}"
47+
implementation "io.grpc:grpc-stub:${versions.grpc}"
48+
runtimeOnly "io.grpc:grpc-all:${versions.grpc}"
49+
runtimeOnly "io.grpc:grpc-protobuf:${versions.grpc}"
50+
runtimeOnly "io.grpc:grpc-protobuf-lite:${versions.grpc}"
51+
runtimeOnly 'io.perfmark:perfmark-api:0.27.0'
52+
runtimeOnly "com.google.guava:failureaccess:1.0.1"
53+
compileOnly "com.google.errorprone:error_prone_annotations:2.31.0"
54+
runtimeOnly('com.google.guava:guava:33.3.1-jre') {
55+
attributes {
56+
attribute(Attribute.of('org.gradle.jvm.environment', String), 'standard-jvm')
57+
}
58+
}
59+
runtimeOnly 'org.apache.parquet:parquet-arrow:1.13.1'
60+
}
61+
62+
tasks.named('test').configure {
63+
jacoco {
64+
excludes = ['org/apache/arrow/flight/**']
65+
}
66+
}
67+
68+
tasks.named('forbiddenApisMain').configure {
69+
replaceSignatureFiles 'jdk-signatures'
70+
71+
excludes = [
72+
'org/apache/arrow/flight/OSFlightServer$Builder.class',
73+
'org/apache/arrow/flight/OSFlightClient$Builder.class',
74+
'org/opensearch/flight/bootstrap/server/ServerConfig$Netty4Configs.class',
75+
'org/opensearch/flight/bootstrap/server/ServerConfig.class',
76+
'org/opensearch/flight/bootstrap/tls/DefaultSslContextProvider.class',
77+
'org/apache/arrow/flight/OpenSearchFlightClient$Builder.class'
78+
]
79+
}
80+
81+
tasks.named('thirdPartyAudit').configure {
82+
ignoreMissingClasses(
83+
'com.google.gson.stream.JsonReader',
84+
'com.google.gson.stream.JsonToken',
85+
'com.google.protobuf.util.Timestamps',
86+
'com.google.rpc.Status',
87+
'com.google.rpc.Status$Builder',
88+
'org.apache.parquet.schema.GroupType',
89+
// Parquet Schema classes
90+
'org.apache.parquet.schema.LogicalTypeAnnotation',
91+
'org.apache.parquet.schema.LogicalTypeAnnotation$DateLogicalTypeAnnotation',
92+
'org.apache.parquet.schema.LogicalTypeAnnotation$DecimalLogicalTypeAnnotation',
93+
'org.apache.parquet.schema.LogicalTypeAnnotation$IntLogicalTypeAnnotation',
94+
'org.apache.parquet.schema.LogicalTypeAnnotation$IntervalLogicalTypeAnnotation',
95+
'org.apache.parquet.schema.LogicalTypeAnnotation$ListLogicalTypeAnnotation',
96+
'org.apache.parquet.schema.LogicalTypeAnnotation$LogicalTypeAnnotationVisitor',
97+
'org.apache.parquet.schema.LogicalTypeAnnotation$StringLogicalTypeAnnotation',
98+
'org.apache.parquet.schema.LogicalTypeAnnotation$TimeLogicalTypeAnnotation',
99+
'org.apache.parquet.schema.LogicalTypeAnnotation$TimeUnit',
100+
'org.apache.parquet.schema.LogicalTypeAnnotation$TimestampLogicalTypeAnnotation',
101+
'org.apache.parquet.schema.MessageType',
102+
'org.apache.parquet.schema.OriginalType',
103+
'org.apache.parquet.schema.PrimitiveType',
104+
'org.apache.parquet.schema.PrimitiveType$PrimitiveTypeName',
105+
'org.apache.parquet.schema.PrimitiveType$PrimitiveTypeNameConverter',
106+
'org.apache.parquet.schema.Type',
107+
'org.apache.parquet.schema.Type$Repetition',
108+
'org.apache.parquet.schema.Types',
109+
'org.apache.parquet.schema.Types$BaseListBuilder',
110+
'org.apache.parquet.schema.Types$GroupBuilder',
111+
'org.apache.parquet.schema.Types$ListBuilder',
112+
'org.apache.parquet.schema.Types$PrimitiveBuilder',
113+
114+
'com.aayushatharva.brotli4j.Brotli4jLoader',
115+
'com.aayushatharva.brotli4j.decoder.DecoderJNI$Status',
116+
'com.aayushatharva.brotli4j.decoder.DecoderJNI$Wrapper',
117+
'com.aayushatharva.brotli4j.encoder.BrotliEncoderChannel',
118+
'com.aayushatharva.brotli4j.encoder.Encoder$Mode',
119+
'com.aayushatharva.brotli4j.encoder.Encoder$Parameters',
120+
// classes are missing
121+
122+
// from io.netty.logging.CommonsLoggerFactory (netty)
123+
'org.apache.commons.logging.Log',
124+
'org.apache.commons.logging.LogFactory',
125+
126+
// from Log4j (deliberate, Netty will fallback to Log4j 2)
127+
'org.apache.log4j.Level',
128+
'org.apache.log4j.Logger',
129+
130+
// from io.netty.handler.ssl.OpenSslEngine (netty)
131+
'io.netty.internal.tcnative.Buffer',
132+
'io.netty.internal.tcnative.CertificateCompressionAlgo',
133+
'io.netty.internal.tcnative.Library',
134+
'io.netty.internal.tcnative.SSL',
135+
'io.netty.internal.tcnative.SSLContext',
136+
'io.netty.internal.tcnative.SSLPrivateKeyMethod',
137+
138+
// from io.netty.handler.ssl.util.BouncyCastleSelfSignedCertGenerator (netty)
139+
'org.bouncycastle.cert.X509v3CertificateBuilder',
140+
'org.bouncycastle.cert.jcajce.JcaX509CertificateConverter',
141+
'org.bouncycastle.operator.jcajce.JcaContentSignerBuilder',
142+
'org.bouncycastle.openssl.PEMEncryptedKeyPair',
143+
'org.bouncycastle.openssl.PEMParser',
144+
'org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter',
145+
'org.bouncycastle.openssl.jcajce.JceOpenSSLPKCS8DecryptorProviderBuilder',
146+
'org.bouncycastle.openssl.jcajce.JcePEMDecryptorProviderBuilder',
147+
'org.bouncycastle.pkcs.PKCS8EncryptedPrivateKeyInfo',
148+
149+
// from io.netty.handler.ssl.JettyNpnSslEngine (netty)
150+
'org.eclipse.jetty.npn.NextProtoNego$ClientProvider',
151+
'org.eclipse.jetty.npn.NextProtoNego$ServerProvider',
152+
'org.eclipse.jetty.npn.NextProtoNego',
153+
154+
// from io.netty.handler.codec.marshalling.ChannelBufferByteInput (netty)
155+
'org.jboss.marshalling.ByteInput',
156+
157+
// from io.netty.handler.codec.marshalling.ChannelBufferByteOutput (netty)
158+
'org.jboss.marshalling.ByteOutput',
159+
160+
// from io.netty.handler.codec.marshalling.CompatibleMarshallingEncoder (netty)
161+
'org.jboss.marshalling.Marshaller',
162+
163+
// from io.netty.handler.codec.marshalling.ContextBoundUnmarshallerProvider (netty)
164+
'org.jboss.marshalling.MarshallerFactory',
165+
'org.jboss.marshalling.MarshallingConfiguration',
166+
'org.jboss.marshalling.Unmarshaller',
167+
168+
'com.google.protobuf.nano.CodedOutputByteBufferNano',
169+
'com.google.protobuf.nano.MessageNano',
170+
'com.ning.compress.BufferRecycler',
171+
'com.ning.compress.lzf.ChunkDecoder',
172+
'com.ning.compress.lzf.ChunkEncoder',
173+
'com.ning.compress.lzf.LZFChunk',
174+
'com.ning.compress.lzf.LZFEncoder',
175+
'com.ning.compress.lzf.util.ChunkDecoderFactory',
176+
'com.ning.compress.lzf.util.ChunkEncoderFactory',
177+
'lzma.sdk.lzma.Encoder',
178+
'net.jpountz.lz4.LZ4Compressor',
179+
'net.jpountz.lz4.LZ4Factory',
180+
'net.jpountz.lz4.LZ4FastDecompressor',
181+
'net.jpountz.xxhash.XXHash32',
182+
'net.jpountz.xxhash.XXHashFactory',
183+
'io.netty.internal.tcnative.AsyncSSLPrivateKeyMethod',
184+
'io.netty.internal.tcnative.AsyncTask',
185+
'io.netty.internal.tcnative.CertificateCallback',
186+
'io.netty.internal.tcnative.CertificateVerifier',
187+
'io.netty.internal.tcnative.ResultCallback',
188+
'io.netty.internal.tcnative.SessionTicketKey',
189+
'io.netty.internal.tcnative.SniHostNameMatcher',
190+
'io.netty.internal.tcnative.SSL',
191+
'io.netty.internal.tcnative.SSLSession',
192+
'io.netty.internal.tcnative.SSLSessionCache',
193+
'org.eclipse.jetty.alpn.ALPN$ClientProvider',
194+
'org.eclipse.jetty.alpn.ALPN$ServerProvider',
195+
'org.eclipse.jetty.alpn.ALPN',
196+
197+
'org.conscrypt.AllocatedBuffer',
198+
'org.conscrypt.BufferAllocator',
199+
'org.conscrypt.Conscrypt',
200+
'org.conscrypt.HandshakeListener',
201+
202+
'reactor.blockhound.BlockHound$Builder',
203+
'reactor.blockhound.integration.BlockHoundIntegration'
204+
)
205+
ignoreViolations(
206+
// Guava internal classes
207+
'com.google.common.cache.Striped64',
208+
'com.google.common.cache.Striped64$1',
209+
'com.google.common.cache.Striped64$Cell',
210+
'com.google.common.hash.LittleEndianByteArray$UnsafeByteArray',
211+
'com.google.common.hash.LittleEndianByteArray$UnsafeByteArray$1',
212+
'com.google.common.hash.LittleEndianByteArray$UnsafeByteArray$2',
213+
'com.google.common.hash.Striped64',
214+
'com.google.common.hash.Striped64$1',
215+
'com.google.common.hash.Striped64$Cell',
216+
'com.google.common.primitives.UnsignedBytes$LexicographicalComparatorHolder$UnsafeComparator',
217+
'com.google.common.primitives.UnsignedBytes$LexicographicalComparatorHolder$UnsafeComparator$1',
218+
'com.google.common.util.concurrent.AbstractFuture$UnsafeAtomicHelper',
219+
'com.google.common.util.concurrent.AbstractFuture$UnsafeAtomicHelper$1',
220+
221+
// Arrow memory classes
222+
'org.apache.arrow.memory.util.MemoryUtil',
223+
'org.apache.arrow.memory.util.MemoryUtil$1',
224+
225+
'io.netty.util.internal.PlatformDependent0',
226+
'io.netty.util.internal.PlatformDependent0$1',
227+
'io.netty.util.internal.PlatformDependent0$2',
228+
'io.netty.util.internal.PlatformDependent0$3',
229+
'io.netty.util.internal.PlatformDependent0$4',
230+
'io.netty.util.internal.PlatformDependent0$6',
231+
'io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueConsumerNodeRef',
232+
'io.netty.util.internal.shaded.org.jctools.queues.BaseLinkedQueueProducerNodeRef',
233+
'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields',
234+
'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields',
235+
'io.netty.util.internal.shaded.org.jctools.queues.BaseMpscLinkedArrayQueueProducerFields',
236+
'io.netty.util.internal.shaded.org.jctools.queues.LinkedQueueNode',
237+
'io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueConsumerIndexField',
238+
'io.netty.util.internal.shaded.org.jctools.queues.MpmcArrayQueueProducerIndexField',
239+
'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueConsumerIndexField',
240+
'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerIndexField',
241+
'io.netty.util.internal.shaded.org.jctools.queues.MpscArrayQueueProducerLimitField',
242+
'io.netty.util.internal.shaded.org.jctools.queues.unpadded.MpscUnpaddedArrayQueueConsumerIndexField',
243+
'io.netty.util.internal.shaded.org.jctools.queues.unpadded.MpscUnpaddedArrayQueueProducerIndexField',
244+
'io.netty.util.internal.shaded.org.jctools.queues.unpadded.MpscUnpaddedArrayQueueProducerLimitField',
245+
'io.netty.util.internal.shaded.org.jctools.util.UnsafeAccess',
246+
'io.netty.util.internal.shaded.org.jctools.util.UnsafeRefArrayAccess',
247+
'io.netty.util.internal.shaded.org.jctools.util.UnsafeLongArrayAccess',
248+
'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator',
249+
'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$1',
250+
'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$2',
251+
'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$3',
252+
'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$4',
253+
'io.netty.handler.ssl.util.OpenJdkSelfSignedCertGenerator$5'
254+
)
255+
}
256+
257+
tasks.named("dependencyLicenses").configure {
258+
mapping from: /netty-.*/, to: 'netty'
259+
}

‎modules/arrow-flight-rpc/licenses/failureaccess-1.0.1.jar.sha1

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
1dcf1de382a0bf95a3d8b0849546c88bac1292c9

0 commit comments

Comments
 (0)