This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new b1da55bfdba Upgrade Reactor Netty security dependencies and preserve
transport behavior (#19828)
b1da55bfdba is described below
commit b1da55bfdba58d62ac1c42e5f1eb07698c37a863
Author: Xiang Fu <[email protected]>
AuthorDate: Wed Oct 14 16:35:47 2026 -0700
Upgrade Reactor Netty security dependencies and preserve transport behavior
(#19828)
* Upgrade Reactor Netty and preserve transport lifecycle behavior
* Use the supported Arrow allocator with Netty 4.2
* Update Netty to 4.2.19 and clarify shutdown registration ordering
Use the latest published 4.2 patch, format the TLS builder chain, and
explain why accepted channels are registered before checking shutdown.
A pre-registration-only check could miss a concurrent shutdown sweep.
Reproduce the ordering race: initialize a channel before shutdown sets
the flag, then register it after the shutdown channel sweep. Keeping the
post-registration check closes this late channel. No lifecycle change
or new test/benchmark is needed.
* Preserve shared Arrow memory limits with Netty 4.2
Reproduction: an Arrow response can exceed a 1 MiB direct-memory ceiling
with the unsafe backend. Preserve caller limits while using the paired shared
allocation factory; release partially allocated encoder vectors on failure.
* Fix ZooKeeper hostname policy with JDK TLS bootstrap
Reproduction: ZooKeeper 3.9.6 with Netty 4.2.19 rejects a trusted
mismatched peer even when zookeeper.ssl.hostnameVerification=false. Preserve
that bootstrap policy through the supported JDK supplier, retaining enabled
verification and custom supplier ownership.
* Consolidate ZooKeeper TLS bootstrap at client creation
* Preserve ZooKeeper TLS bootstrap and cipher compatibility
---
.../apache/pinot/client/DynamicBrokerSelector.java | 2 +
pinot-common/pom.xml | 2 +-
.../response/encoder/ArrowResponseEncoder.java | 25 ++-
.../apache/pinot/common/utils/ArrowAllocators.java | 143 ++++++++++++++
.../pinot/common/utils/ServiceStartableUtils.java | 2 +
.../pinot/common/utils/helix/HelixHelper.java | 11 ++
.../apache/pinot/common/utils/tls/TlsUtils.java | 7 +-
.../utils/tls/ZooKeeperSslContextSupplier.java | 77 ++++++++
.../pinot/common/utils/ArrowAllocatorsTest.java | 137 ++++++++++++++
.../utils/tls/ZooKeeperSslContextSupplierTest.java | 209 +++++++++++++++++++++
.../helix/core/util/HelixSetupUtils.java | 2 +
.../apache/pinot/core/transport/QueryServer.java | 11 ++
.../pinot-input-format/pinot-arrow/pom.xml | 7 +-
.../inputformat/arrow/ArrowMessageDecoder.java | 3 +-
.../inputformat/arrow/ArrowRecordReader.java | 3 +-
.../inputformat/arrow/BatchedArrowFileSource.java | 3 +-
.../server/predownload/PredownloadZKClient.java | 2 +
.../org/apache/pinot/tools/PinotToolLauncher.java | 2 +
.../pinot/tools/admin/PinotAdministrator.java | 2 +
.../pinot/tools/perf/PerfBenchmarkDriver.java | 5 +
pom.xml | 2 +-
21 files changed, 646 insertions(+), 11 deletions(-)
diff --git
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/DynamicBrokerSelector.java
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/DynamicBrokerSelector.java
index 5c8622149ba..f4cf5c03b04 100644
---
a/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/DynamicBrokerSelector.java
+++
b/pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/DynamicBrokerSelector.java
@@ -33,6 +33,7 @@ import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;
import org.apache.pinot.client.utils.BrokerSelectorUtils;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -68,6 +69,7 @@ public class DynamicBrokerSelector implements BrokerSelector,
IZkDataListener {
@VisibleForTesting
protected ZkClient getZkClient(String zkServers) {
+ ZooKeeperSslContextSupplier.initialize();
return new ZkClient(zkServers);
}
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index 050a315bec7..0c5fc3b1494 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -201,7 +201,7 @@
</dependency>
<dependency>
<groupId>org.apache.arrow</groupId>
- <artifactId>arrow-memory-netty</artifactId>
+ <artifactId>arrow-memory-unsafe</artifactId>
</dependency>
<dependency>
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/encoder/ArrowResponseEncoder.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/encoder/ArrowResponseEncoder.java
index 4fb37bb9776..6b2e9926528 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/encoder/ArrowResponseEncoder.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/encoder/ArrowResponseEncoder.java
@@ -46,12 +46,13 @@ import org.apache.arrow.vector.types.pojo.FieldType;
import org.apache.arrow.vector.types.pojo.Schema;
import org.apache.arrow.vector.util.JsonStringArrayList;
import org.apache.pinot.common.response.broker.ResultTable;
+import org.apache.pinot.common.utils.ArrowAllocators;
import org.apache.pinot.common.utils.DataSchema;
import org.apache.pinot.spi.utils.MapUtils;
public class ArrowResponseEncoder implements ResponseEncoder {
- private static final RootAllocator ALLOCATOR = new
RootAllocator(Long.MAX_VALUE);
+ private static final RootAllocator ALLOCATOR =
ArrowAllocators.newRootAllocator(Long.MAX_VALUE);
@Override
public byte[] encodeResultTable(ResultTable resultTable, int startRow, int
length)
@@ -69,8 +70,26 @@ public class ArrowResponseEncoder implements ResponseEncoder
{
private VectorSchemaRoot createVectorSchemaRoot(ResultTable resultTable,
DataSchema schema, int startRow,
int length) {
- List<Field> fields = new ArrayList<>();
List<FieldVector> vectors = new ArrayList<>();
+ try {
+ return createVectorSchemaRoot(resultTable, schema, startRow, length,
vectors);
+ } catch (RuntimeException | Error e) {
+ for (FieldVector vector : vectors) {
+ try {
+ vector.close();
+ } catch (RuntimeException | Error closeFailure) {
+ if (closeFailure != e) {
+ e.addSuppressed(closeFailure);
+ }
+ }
+ }
+ throw e;
+ }
+ }
+
+ private VectorSchemaRoot createVectorSchemaRoot(ResultTable resultTable,
DataSchema schema, int startRow,
+ int length, List<FieldVector> vectors) {
+ List<Field> fields = new ArrayList<>();
int numColumns = schema.getColumnNames().length; // assuming getter
returns String[]
// Create an Arrow Field and vector for each column based on its type.
@@ -196,8 +215,8 @@ public class ArrowResponseEncoder implements
ResponseEncoder {
throw new UnsupportedOperationException("Unsupported column type: "
+ colType);
}
fields.add(field);
- vector.allocateNew();
vectors.add(vector);
+ vector.allocateNew();
}
// Determine the actual end row.
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ArrowAllocators.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ArrowAllocators.java
new file mode 100644
index 00000000000..d3ef1f2d9ec
--- /dev/null
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ArrowAllocators.java
@@ -0,0 +1,143 @@
+/**
+ * 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.pinot.common.utils;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.util.internal.PlatformDependent;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.Nullable;
+import org.apache.arrow.memory.AllocationManager;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.OutOfMemoryException;
+import org.apache.arrow.memory.ReferenceManager;
+import org.apache.arrow.memory.RootAllocator;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+
+/// Creates Arrow allocators with a shared allocation budget and Netty-backed,
int-sized buffers.
+///
+/// The budget uses Netty's existing direct-memory ceiling. Int-sized
allocations also participate in
+/// Netty's direct-memory counter, preserving the limit shared with transport
buffers without relying
+/// on Arrow 19's private Netty buffer casts. Larger allocations retain
Arrow's raw-memory path but
+/// now count against the shared Arrow budget; that path is separate from
Netty's counter.
+///
+/// Callers still own their root allocator and retain their existing
per-allocator limit. Children
+/// inherit the allocation manager. Allocations through other factories or
native Arrow libraries
+/// are outside this budget.
+public final class ArrowAllocators {
+ private static final long MAX_ALLOCATION =
PlatformDependent.maxDirectMemory();
+ private static final AtomicLong ALLOCATED_MEMORY = new AtomicLong();
+ private static final ArrowBuf EMPTY_BUFFER = new
ArrowBuf(ReferenceManager.NO_OP, null, 0, 0);
+ private static final AllocationManager.Factory ALLOCATION_MANAGER_FACTORY =
new AllocationManager.Factory() {
+ @Override
+ public AllocationManager create(BufferAllocator allocator, long size) {
+ reserve(size);
+ try {
+ return new NettyAllocationManager(allocator, size);
+ } catch (RuntimeException | Error e) {
+ ALLOCATED_MEMORY.addAndGet(-size);
+ if (e instanceof OutOfMemoryError) {
+ throw new OutOfMemoryException("Failure allocating buffer.", e);
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public ArrowBuf empty() {
+ return EMPTY_BUFFER;
+ }
+ };
+
+ private ArrowAllocators() {
+ }
+
+ /// Creates a caller-owned root with the supplied accounting limit and the
shared physical allocation budget.
+ public static RootAllocator newRootAllocator(long maxAllocation) {
+ return new
RootAllocator(RootAllocator.configBuilder().maxAllocation(maxAllocation)
+ .allocationManagerFactory(ALLOCATION_MANAGER_FACTORY).build());
+ }
+
+ private static void reserve(long size) {
+ checkArgument(size >= 0, "Allocation size must be non-negative: %s", size);
+ long allocated;
+ do {
+ allocated = ALLOCATED_MEMORY.get();
+ if (size > MAX_ALLOCATION - allocated) {
+ throw new OutOfMemoryException("Arrow allocation of " + size + " bytes
exceeds the shared limit of "
+ + MAX_ALLOCATION + " bytes (already allocated: " + allocated +
").");
+ }
+ } while (!ALLOCATED_MEMORY.compareAndSet(allocated, allocated + size));
+ }
+
+ @VisibleForTesting
+ static long getAllocatedMemory() {
+ return ALLOCATED_MEMORY.get();
+ }
+
+ /// Owns one physical allocation until the final Arrow reference is released.
+ private static final class NettyAllocationManager extends AllocationManager {
+ private final long _size;
+ private final long _address;
+ @Nullable
+ private final ByteBuf _buffer;
+
+ private NettyAllocationManager(BufferAllocator allocator, long size) {
+ super(allocator);
+ _size = size;
+ if (size > Integer.MAX_VALUE) {
+ _buffer = null;
+ _address = PlatformDependent.allocateMemory(size);
+ } else {
+ ByteBuf buffer = PooledByteBufAllocator.DEFAULT.directBuffer((int)
size, (int) size);
+ try {
+ _address = buffer.memoryAddress();
+ } catch (RuntimeException | Error e) {
+ buffer.release();
+ throw e;
+ }
+ _buffer = buffer;
+ }
+ }
+
+ @Override
+ public long getSize() {
+ return _size;
+ }
+
+ @Override
+ protected long memoryAddress() {
+ return _address;
+ }
+
+ @Override
+ protected void release0() {
+ if (_buffer != null) {
+ _buffer.release();
+ } else {
+ PlatformDependent.freeMemory(_address);
+ }
+ ALLOCATED_MEMORY.addAndGet(-_size);
+ }
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
index a3fcb7b68ff..ff7c61a7e0b 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/ServiceStartableUtils.java
@@ -27,6 +27,7 @@ import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.pinot.common.utils.request.RequestUtils;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.segment.spi.index.ForwardIndexConfig;
import org.apache.pinot.spi.config.table.FieldConfig.CompressionCodec;
import org.apache.pinot.spi.data.FieldSpec;
@@ -56,6 +57,7 @@ public class ServiceStartableUtils {
/// - pinot.all.\* will be replaced to role specific config, e.g.
pinot.controller.\* for controllers
public static void applyClusterConfig(PinotConfiguration instanceConfig,
String zkAddress, String clusterName,
ServiceRole serviceRole) {
+ ZooKeeperSslContextSupplier.initialize();
int zkClientSessionConfig =
instanceConfig.getProperty(CommonConstants.Helix.ZkClient.ZK_CLIENT_SESSION_TIMEOUT_MS_CONFIG,
CommonConstants.Helix.ZkClient.DEFAULT_SESSION_TIMEOUT_MS);
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java
index 8377938ae22..e8636bea991 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/helix/HelixHelper.java
@@ -33,6 +33,8 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
import org.apache.helix.PropertyKey.Builder;
import org.apache.helix.model.ExternalView;
import org.apache.helix.model.HelixConfigScope;
@@ -46,6 +48,7 @@ import
org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.pinot.common.helix.ExtraInstanceConfig;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.utils.config.TagNameUtils;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.common.version.PinotVersion;
import org.apache.pinot.common.workload.WorkloadChangeListener;
import org.apache.pinot.spi.config.table.TableConfig;
@@ -78,6 +81,14 @@ public class HelixHelper {
public static final String BROKER_RESOURCE =
CommonConstants.Helix.BROKER_RESOURCE_INSTANCE;
+ /// Creates a ZooKeeper-backed manager with Pinot's TLS bootstrap policy
applied before construction.
+ /// Callers retain responsibility for the manager's connect/disconnect
lifecycle.
+ public static HelixManager createZkHelixManager(String clusterName, String
instanceName, InstanceType instanceType,
+ String zkAddress) {
+ ZooKeeperSslContextSupplier.initialize();
+ return HelixManagerFactory.getZKHelixManager(clusterName, instanceName,
instanceType, zkAddress);
+ }
+
public static IdealState cloneIdealState(IdealState idealState) {
return new IdealState(
(ZNRecord)
ZN_RECORD_SERIALIZER.deserialize(ZN_RECORD_SERIALIZER.serialize(idealState.getRecord())));
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/TlsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/TlsUtils.java
index b5121befa4a..118d212dc7d 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/TlsUtils.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/TlsUtils.java
@@ -342,8 +342,11 @@ public final class TlsUtils {
SSLFactory sslFactory =
RenewableTlsUtils.createSSLFactoryAndEnableAutoRenewalWhenUsingFileStores(
tlsConfig, PinotInsecureMode::isPinotInInsecureMode);
- SslContextBuilder sslContextBuilder =
-
SslContextBuilder.forClient().sslProvider(SslProvider.valueOf(tlsConfig.getSslProvider()));
+ // Keep Pinot's existing TLS behavior: verify certificate trust without
hostname verification.
+ // Netty 4.2 enables hostname verification by default, which requires a
peer hostname.
+ SslContextBuilder sslContextBuilder = SslContextBuilder.forClient()
+ .sslProvider(SslProvider.valueOf(tlsConfig.getSslProvider()))
+ .endpointIdentificationAlgorithm(null);
sslFactory.getKeyManagerFactory().ifPresent(sslContextBuilder::keyManager);
sslFactory.getTrustManagerFactory().ifPresent(sslContextBuilder::trustManager);
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/ZooKeeperSslContextSupplier.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/ZooKeeperSslContextSupplier.java
new file mode 100644
index 00000000000..4ef7a6eec13
--- /dev/null
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/tls/ZooKeeperSslContextSupplier.java
@@ -0,0 +1,77 @@
+/**
+ * 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.pinot.common.utils.tls;
+
+import io.netty.handler.ssl.CipherSuiteConverter;
+import java.util.Properties;
+import java.util.function.Supplier;
+import javax.net.ssl.SSLContext;
+import org.apache.zookeeper.common.ClientX509Util;
+import org.apache.zookeeper.common.X509Exception.SSLContextException;
+import org.apache.zookeeper.common.ZKConfig;
+
+
+/// Preserves disabled ZooKeeper hostname verification with Netty 4.2 using
ZooKeeper's JDK context supplier hook.
+///
+/// Pinot's Helix clients use process-wide ZooKeeper TLS bootstrap properties.
Call [#initialize()] before creating
+/// these clients; existing custom suppliers retain ownership of their context
and provider. The vendor's no-argument
+/// supplier cannot consume arbitrary per-client TLS overrides, so it is
intended for these bootstrap-owned clients.
+/// Each context uses ZooKeeper's configured stores, protocol, revocation and
FIPS policy, without caching key material.
+public class ZooKeeperSslContextSupplier implements Supplier<SSLContext> {
+ /// Installs the JDK adapter only when bootstrap configuration explicitly
disables hostname verification.
+ /// Enabled verification and custom suppliers keep their existing
construction paths. TLS bootstrap configuration
+ /// must be supplied before client creation; it cannot be obtained from the
cluster being connected to.
+ public static void initialize() {
+ Properties properties = System.getProperties();
+ synchronized (properties) {
+ try (ClientX509Util util = new ClientX509Util()) {
+ ZKConfig config = new ZKConfig();
+ if (util.isServerHostnameVerificationEnabled(config)
+ || config.getProperty(util.getSslContextSupplierClassProperty())
!= null) {
+ return;
+ }
+ // The supplied JDK context needs JSSE names for ciphers previously
configured for OpenSSL.
+ String cipherSuites =
config.getProperty(util.getSslCipherSuitesProperty());
+ if (cipherSuites != null) {
+ String[] ciphers = cipherSuites.split(",", -1);
+ for (int i = 0; i < ciphers.length; i++) {
+ String javaCipher = CipherSuiteConverter.toJava(ciphers[i], "TLS");
+ if (javaCipher != null) {
+ ciphers[i] = javaCipher;
+ }
+ }
+ properties.setProperty(util.getSslCipherSuitesProperty(),
String.join(",", ciphers));
+ }
+ // ZooKeeper 3.9.6 supports supplied SSLContext instances only with
its JDK provider.
+ properties.setProperty(util.getSslProviderProperty(), "JDK");
+ properties.setProperty(util.getSslContextSupplierClassProperty(),
ZooKeeperSslContextSupplier.class.getName());
+ }
+ }
+ }
+
+ @Override
+ public SSLContext get() {
+ try (ClientX509Util util = new ClientX509Util()) {
+ // Bypass supplier loading without temporarily removing the process-wide
supplier property.
+ return util.createSSLContextAndOptionsFromConfig(new
ZKConfig()).getSSLContext();
+ } catch (SSLContextException e) {
+ throw new IllegalStateException("Unable to create the ZooKeeper TLS
context from bootstrap configuration", e);
+ }
+ }
+}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/ArrowAllocatorsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ArrowAllocatorsTest.java
new file mode 100644
index 00000000000..d56d818a111
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/ArrowAllocatorsTest.java
@@ -0,0 +1,137 @@
+/**
+ * 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.pinot.common.utils;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.OutOfMemoryException;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.pinot.common.response.broker.ResultTable;
+import org.apache.pinot.common.response.encoder.ArrowResponseEncoder;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Verifies allocation limits and encoder recovery in an isolated JVM.
+public class ArrowAllocatorsTest {
+ @Test
+ public void testProcessMemoryLimitAndEncoderRecovery()
+ throws Exception {
+ String classpath = System.getProperty("surefire.test.class.path",
System.getProperty("java.class.path"));
+ Process process = new
ProcessBuilder(Path.of(System.getProperty("java.home"), "bin",
"java").toString(),
+ "-Xmx64m", "-XX:MaxDirectMemorySize=1m",
"--add-opens=java.base/java.nio=ALL-UNNAMED",
+ "-Dio.netty.maxDirectMemory=1048576",
"-Dio.netty.tryReflectionSetAccessible=true",
+ "-Dio.netty.allocator.numDirectArenas=1",
"-Dio.netty.allocator.maxOrder=4", "-cp", classpath,
+ MemoryLimitProbe.class.getName()).redirectErrorStream(true).start();
+ try {
+ assertTrue(process.waitFor(30, TimeUnit.SECONDS), "Arrow memory limit
probe timed out");
+ String output = new String(process.getInputStream().readAllBytes(),
StandardCharsets.UTF_8);
+ assertEquals(process.exitValue(), 0, output);
+ } finally {
+ process.destroyForcibly();
+ }
+ }
+
+ /// Exercises the production allocator and encoder in a JVM with a bounded
direct-memory limit.
+ public static final class MemoryLimitProbe {
+ private MemoryLimitProbe() {
+ }
+
+ public static void main(String[] args)
+ throws Exception {
+ testSharedBudgetAndChildren();
+ testNettyBudgetAndRetainedReferences();
+ testEncoderRecovery();
+ }
+
+ private static void testSharedBudgetAndChildren() {
+ try (RootAllocator first =
ArrowAllocators.newRootAllocator(Long.MAX_VALUE);
+ RootAllocator second =
ArrowAllocators.newRootAllocator(Long.MAX_VALUE);
+ BufferAllocator child = first.newChildAllocator("query", 0,
Long.MAX_VALUE)) {
+ ArrowBuf firstBuffer = child.buffer(512 * 1024);
+ ArrowBuf secondBuffer = second.buffer(256 * 1024);
+ try {
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 768 * 1024L);
+ expectThrows(OutOfMemoryException.class, () -> second.buffer(512 *
1024));
+ assertEquals(second.getAllocatedMemory(), 256 * 1024L);
+ firstBuffer.close();
+ firstBuffer = null;
+ try (ArrowBuf recovered = second.buffer(512 * 1024)) {
+ recovered.setLong(0, 42L);
+ assertEquals(recovered.getLong(0), 42L);
+ assertEquals(recovered.memoryAddress() % Long.BYTES, 0L);
+ }
+ expectThrows(OutOfMemoryException.class, () -> child.buffer((long)
Integer.MAX_VALUE + 1));
+ } finally {
+ if (firstBuffer != null) {
+ firstBuffer.close();
+ }
+ secondBuffer.close();
+ }
+ }
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 0L);
+ }
+
+ private static void testNettyBudgetAndRetainedReferences() {
+ try (RootAllocator allocator =
ArrowAllocators.newRootAllocator(Long.MAX_VALUE)) {
+ ByteBuf occupied = PooledByteBufAllocator.DEFAULT.directBuffer(1024 *
1024, 1024 * 1024);
+ try {
+ expectThrows(OutOfMemoryException.class, () ->
allocator.buffer(4096));
+ assertEquals(allocator.getAllocatedMemory(), 0L);
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 0L);
+ } finally {
+ occupied.release();
+ }
+ ArrowBuf buffer = allocator.buffer(512 * 1024);
+ buffer.getReferenceManager().retain();
+ buffer.close();
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 512 * 1024L);
+ buffer.close();
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 0L);
+ }
+ }
+
+ private static void testEncoderRecovery()
+ throws Exception {
+ ArrowResponseEncoder encoder = new ArrowResponseEncoder();
+ DataSchema schema = new DataSchema(new String[]{"id", "value"},
+ new ColumnDataType[]{ColumnDataType.INT, ColumnDataType.STRING});
+ ResultTable oversized = new ResultTable(schema,
Collections.singletonList(new Object[]{1,
+ "x".repeat(2 * 1024 * 1024)}));
+ expectThrows(OutOfMemoryException.class, () ->
encoder.encodeResultTable(oversized, 0, 1));
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 0L);
+ ResultTable small = new ResultTable(schema,
Collections.singletonList(new Object[]{1, "ok"}));
+ byte[] encoded = encoder.encodeResultTable(small, 0, 1);
+ ResultTable decoded = encoder.decodeResultTable(encoded, 1, schema);
+ assertEquals(decoded.getRows().size(), 1);
+ assertEquals(decoded.getRows().get(0), small.getRows().get(0));
+ assertEquals(ArrowAllocators.getAllocatedMemory(), 0L);
+ }
+ }
+}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/tls/ZooKeeperSslContextSupplierTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/tls/ZooKeeperSslContextSupplierTest.java
new file mode 100644
index 00000000000..ace04107db2
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/tls/ZooKeeperSslContextSupplierTest.java
@@ -0,0 +1,209 @@
+/**
+ * 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.pinot.common.utils.tls;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.ssl.SslContext;
+import io.netty.handler.ssl.SslHandler;
+import java.io.InputStream;
+import java.net.InetSocketAddress;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.KeyStore;
+import java.security.cert.CertificateException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLServerSocket;
+import javax.net.ssl.SSLSocket;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.pinot.common.utils.helix.HelixHelper;
+import org.apache.zookeeper.common.ClientX509Util;
+import org.apache.zookeeper.common.ZKConfig;
+import org.mockito.MockedStatic;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.mockStatic;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Exercises the released ZooKeeper context factory against a trusted
certificate with a mismatched peer identity.
+@Test(singleThreaded = true)
+public class ZooKeeperSslContextSupplierTest {
+ private final Map<String, String> _savedProperties = new HashMap<>();
+
+ @BeforeMethod
+ public void saveBootstrapProperties() {
+ try (ClientX509Util util = new ClientX509Util()) {
+ for (String key : new
String[]{util.getSslContextSupplierClassProperty(),
util.getSslProviderProperty(),
+ util.getSslHostnameVerificationEnabledProperty(),
util.getFipsModeProperty(),
+ util.getSslCipherSuitesProperty(),
util.getSslTruststoreLocationProperty(),
+ util.getSslTruststorePasswdProperty(),
util.getSslTruststoreTypeProperty()}) {
+ _savedProperties.put(key, System.getProperty(key));
+ }
+ System.clearProperty(util.getSslContextSupplierClassProperty());
+ }
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void restoreBootstrapProperties() {
+ _savedProperties.forEach((key, value) -> {
+ if (value == null) {
+ System.clearProperty(key);
+ } else {
+ System.setProperty(key, value);
+ }
+ });
+ _savedProperties.clear();
+ }
+
+ @Test
+ public void testCustomSupplierKeepsProviderOwnership() {
+ try (ClientX509Util util = new ClientX509Util();
+ MockedStatic<HelixManagerFactory> ignored =
mockStatic(HelixManagerFactory.class)) {
+ System.setProperty(util.getSslHostnameVerificationEnabledProperty(),
"false");
+ System.setProperty(util.getSslContextSupplierClassProperty(),
"custom.supplier");
+ System.setProperty(util.getSslProviderProperty(), "OPENSSL");
+ System.setProperty(util.getSslCipherSuitesProperty(),
"ECDHE-RSA-AES128-GCM-SHA256");
+ HelixHelper.createZkHelixManager("test-cluster", "test-instance",
InstanceType.SPECTATOR, "localhost:2181");
+
assertEquals(System.getProperty(util.getSslContextSupplierClassProperty()),
"custom.supplier");
+ assertEquals(System.getProperty(util.getSslProviderProperty()),
"OPENSSL");
+ assertEquals(System.getProperty(util.getSslCipherSuitesProperty()),
"ECDHE-RSA-AES128-GCM-SHA256");
+ }
+ }
+
+ @DataProvider
+ public Object[][] hostnamePolicies() {
+ return new Object[][]{{false, false, "OPENSSL"}, {false, true,
"OPENSSL_REFCNT"},
+ {true, false, "JDK"}, {true, true, "JDK"}};
+ }
+
+ @Test(dataProvider = "hostnamePolicies")
+ public void testConfiguredHostnamePolicy(boolean verifyHostname, boolean
fips, String provider)
+ throws Exception {
+ try (ClientX509Util util = new ClientX509Util();
+ MockedStatic<HelixManagerFactory> factory =
mockStatic(HelixManagerFactory.class)) {
+ System.setProperty(util.getSslHostnameVerificationEnabledProperty(),
Boolean.toString(verifyHostname));
+ System.setProperty(util.getFipsModeProperty(), Boolean.toString(fips));
+ System.setProperty(util.getSslProviderProperty(), provider);
+ System.setProperty(util.getSslCipherSuitesProperty(),
+ verifyHostname ? "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256" :
"ECDHE-RSA-AES128-GCM-SHA256");
+ System.setProperty(util.getSslTruststoreLocationProperty(),
resourcePath("truststore.p12").toString());
+ System.setProperty(util.getSslTruststorePasswdProperty(), "changeit");
+ System.setProperty(util.getSslTruststoreTypeProperty(), "PKCS12");
+ AtomicReference<SslContext> context = new AtomicReference<>();
+ // Helix reads cloud config during manager construction. Capture that
first client's context before returning.
+ factory.when(() -> HelixManagerFactory.getZKHelixManager("test-cluster",
"test-instance", InstanceType.SPECTATOR,
+ "localhost:2181"))
+ .thenAnswer(invocation -> {
+ context.set(util.createNettySslContextForClient(new ZKConfig()));
+ return null;
+ });
+ HelixHelper.createZkHelixManager("test-cluster", "test-instance",
InstanceType.SPECTATOR, "localhost:2181");
+ SslContext clientContext = context.get();
+
+ KeyStore keys = KeyStore.getInstance("PKCS12");
+ try (InputStream stream =
Files.newInputStream(resourcePath("keystore.p12"))) {
+ keys.load(stream, "changeit".toCharArray());
+ }
+ KeyManagerFactory keyManager =
KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+ keyManager.init(keys, "changeit".toCharArray());
+ SSLContext serverContext = SSLContext.getInstance("TLS");
+ serverContext.init(keyManager.getKeyManagers(), null, null);
+ EventLoopGroup group = new MultiThreadIoEventLoopGroup(1,
NioIoHandler.newFactory());
+ Channel channel = null;
+ try (SSLServerSocket server = (SSLServerSocket)
serverContext.getServerSocketFactory().createServerSocket()) {
+ server.bind(new InetSocketAddress("127.0.0.1", 0));
+ server.setSoTimeout(5000);
+ int port = server.getLocalPort();
+ SslHandler ssl = clientContext.newHandler(ByteBufAllocator.DEFAULT,
"127.0.0.1", port);
+ Thread worker = new Thread(() -> {
+ try (SSLSocket socket = (SSLSocket) server.accept()) {
+ socket.setSoTimeout(5000);
+ socket.startHandshake();
+ } catch (Exception e) {
+ // An enabled-policy client rejects this deliberately mismatched
certificate and closes its connection.
+ }
+ }, "zookeeper-tls-test-server");
+ worker.setDaemon(true);
+ worker.start();
+ try {
+ channel = new
Bootstrap().group(group).channel(NioSocketChannel.class)
+ .handler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel socket) {
+ ssl.setHandshakeTimeoutMillis(5000);
+ socket.pipeline().addLast(ssl);
+ }
+ }).connect("127.0.0.1", port).sync().channel();
+ assertTrue(ssl.handshakeFuture().await(6000), "TLS handshake did not
complete");
+ if (verifyHostname) {
+ assertFalse(ssl.handshakeFuture().isSuccess());
+ assertTrue(hasCertificateFailure(ssl.handshakeFuture().cause()),
+ "Expected a certificate rejection, not a connection or timeout
failure");
+ } else {
+ assertTrue(ssl.handshakeFuture().isSuccess(),
String.valueOf(ssl.handshakeFuture().cause()));
+ assertEquals(ssl.engine().getSession().getCipherSuite(),
+ "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256");
+ }
+ } finally {
+ if (channel != null) {
+ channel.close().sync();
+ }
+ server.close();
+ worker.join(6000);
+ assertFalse(worker.isAlive(), "TLS fixture did not stop");
+ }
+ } finally {
+ group.shutdownGracefully(0, 0, TimeUnit.SECONDS).sync();
+ }
+ }
+ }
+
+ private static boolean hasCertificateFailure(Throwable failure) {
+ for (Throwable cause = failure; cause != null; cause = cause.getCause()) {
+ if (cause instanceof CertificateException) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static Path resourcePath(String name)
+ throws Exception {
+ return
Path.of(ZooKeeperSslContextSupplierTest.class.getClassLoader().getResource("tls/"
+ name).toURI());
+ }
+}
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
index 832ce2346de..f5847028c1e 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/util/HelixSetupUtils.java
@@ -45,6 +45,7 @@ import
org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.pinot.common.utils.ZkStarter;
import org.apache.pinot.common.utils.helix.LeadControllerUtils;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.controller.ControllerConf;
import
org.apache.pinot.controller.helix.core.PinotHelixBrokerResourceOnlineOfflineStateModelGenerator;
import
org.apache.pinot.controller.helix.core.PinotHelixSegmentOnlineOfflineStateModelGenerator;
@@ -68,6 +69,7 @@ public class HelixSetupUtils {
/// the config doesn't already exist.
public static void setupHelixClusterWithDefaultConfigs(String zkAddress,
String clusterName,
Map<String, String> defaultConfigs) {
+ ZooKeeperSslContextSupplier.initialize();
HelixAdmin admin = new
ZKHelixAdmin.Builder().setZkAddress(zkAddress).build();
try {
if (admin.getClusters().contains(clusterName)) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryServer.java
b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryServer.java
index 51355be511f..c51fa77c5e0 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryServer.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/transport/QueryServer.java
@@ -62,6 +62,7 @@ public class QueryServer {
private final ChannelHandler _instanceRequestHandler;
private ServerSocketChannel _channel;
private final ConcurrentHashMap<SocketChannel, Boolean> _allChannels = new
ConcurrentHashMap<>();
+ private volatile boolean _shuttingDown;
/// Create an unsecured server instance
@@ -135,8 +136,13 @@ public class QueryServer {
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
+ // Register first so shutdown sees this channel, or the check
below closes a late registration.
_allChannels.put(ch, true);
ch.closeFuture().addListener(f -> _allChannels.remove(ch));
+ if (_shuttingDown) {
+ ch.close();
+ return;
+ }
ch.pipeline()
.addLast(ChannelHandlerFactory.getDirectOOMHandler(null,
null, null, _allChannels, _channel));
@@ -160,8 +166,13 @@ public class QueryServer {
}
public void shutDown() {
+ _shuttingDown = true;
try {
_channel.close().sync();
+ // Close accepted connections explicitly, including registrations racing
with listener shutdown.
+ for (SocketChannel channel : _allChannels.keySet()) {
+ channel.close();
+ }
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
diff --git a/pinot-plugins/pinot-input-format/pinot-arrow/pom.xml
b/pinot-plugins/pinot-input-format/pinot-arrow/pom.xml
index 71ce5a6db8f..ee9a5ec4a9c 100644
--- a/pinot-plugins/pinot-input-format/pinot-arrow/pom.xml
+++ b/pinot-plugins/pinot-input-format/pinot-arrow/pom.xml
@@ -36,6 +36,11 @@
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.arrow</groupId>
<artifactId>arrow-compression</artifactId>
@@ -50,7 +55,7 @@
</dependency>
<dependency>
<groupId>org.apache.arrow</groupId>
- <artifactId>arrow-memory-netty</artifactId>
+ <artifactId>arrow-memory-unsafe</artifactId>
</dependency>
<dependency>
<groupId>org.apache.arrow</groupId>
diff --git
a/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowMessageDecoder.java
b/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowMessageDecoder.java
index dfd80262b2d..d153e26b8b6 100644
---
a/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowMessageDecoder.java
+++
b/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowMessageDecoder.java
@@ -31,6 +31,7 @@ import javax.annotation.Nullable;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowStreamReader;
+import org.apache.pinot.common.utils.ArrowAllocators;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
import org.apache.pinot.spi.plugin.PluginManager;
@@ -84,7 +85,7 @@ public class ArrowMessageDecoder implements
StreamMessageDecoder<byte[]> {
// Initialize Arrow allocator with configurable memory limit
long allocatorLimit =
Long.parseLong(props.getOrDefault(ARROW_ALLOCATOR_LIMIT,
DEFAULT_ALLOCATOR_LIMIT));
- _allocator = new RootAllocator(allocatorLimit);
+ _allocator = ArrowAllocators.newRootAllocator(allocatorLimit);
LOGGER.info("Initialized ArrowMessageDecoder for topic: {} with allocator
limit: {} bytes", topicName,
allocatorLimit);
diff --git
a/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowRecordReader.java
b/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowRecordReader.java
index dbd453201fb..91b0278128b 100644
---
a/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowRecordReader.java
+++
b/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/ArrowRecordReader.java
@@ -26,6 +26,7 @@ import javax.annotation.Nullable;
import org.apache.arrow.memory.RootAllocator;
import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowFileReader;
+import org.apache.pinot.common.utils.ArrowAllocators;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordReader;
import org.apache.pinot.spi.data.readers.RecordReaderConfig;
@@ -56,7 +57,7 @@ public class ArrowRecordReader implements RecordReader {
}
_extractor.init(fieldsToRead, extractorConfig);
_dataFile = dataFile;
- _allocator = new RootAllocator(allocatorLimit);
+ _allocator = ArrowAllocators.newRootAllocator(allocatorLimit);
openFile();
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/BatchedArrowFileSource.java
b/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/BatchedArrowFileSource.java
index fab615b7394..e67fc22433e 100644
---
a/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/BatchedArrowFileSource.java
+++
b/pinot-plugins/pinot-input-format/pinot-arrow/src/main/java/org/apache/pinot/plugin/inputformat/arrow/BatchedArrowFileSource.java
@@ -39,6 +39,7 @@ import org.apache.arrow.vector.ipc.ArrowFileReader;
import org.apache.arrow.vector.ipc.message.ArrowBlock;
import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.pinot.common.utils.ArrowAllocators;
/// Reads an Apache Arrow IPC file **one record batch at a time** through the
seekable
@@ -76,7 +77,7 @@ final class BatchedArrowFileSource implements Closeable {
BatchedArrowFileSource(File dataFile, long allocatorLimit, boolean
extractRawTimeValues)
throws IOException {
- _allocator = new RootAllocator(allocatorLimit);
+ _allocator = ArrowAllocators.newRootAllocator(allocatorLimit);
_extractRawTimeValues = extractRawTimeValues;
boolean initialized = false;
try {
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
index 5defca1d3b1..31da2e1eee7 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/predownload/PredownloadZKClient.java
@@ -41,6 +41,7 @@ import
org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.impl.factory.SharedZkClientFactory;
import org.apache.pinot.common.metadata.ZKMetadataProvider;
import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.Schema;
@@ -71,6 +72,7 @@ public class PredownloadZKClient implements AutoCloseable {
}
public void start() {
+ ZooKeeperSslContextSupplier.initialize();
RealmAwareZkClient.RealmAwareZkClientConfig config = new
RealmAwareZkClient.RealmAwareZkClientConfig();
config.setConnectInitTimeout(ZK_CONNECTION_TIMEOUT_MS);
config.setZkSerializer(new ZNRecordSerializer());
diff --git
a/pinot-tools/src/main/java/org/apache/pinot/tools/PinotToolLauncher.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/PinotToolLauncher.java
index b1c27a9b90d..171f9268011 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/PinotToolLauncher.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/PinotToolLauncher.java
@@ -20,6 +20,7 @@ package org.apache.pinot.tools;
import java.util.HashMap;
import java.util.Map;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.tools.filesystem.PinotFSBenchmarkRunner;
import org.apache.pinot.tools.perf.PerfBenchmarkRunner;
@@ -34,6 +35,7 @@ public class PinotToolLauncher {
private static final Map<String, Command> SUBCOMMAND_MAP = new HashMap<>();
static {
+ ZooKeeperSslContextSupplier.initialize();
SUBCOMMAND_MAP.put("UpdateSegmentState", new UpdateSegmentState());
SUBCOMMAND_MAP.put("AutoAddInvertedIndex", new AutoAddInvertedIndexTool());
SUBCOMMAND_MAP.put("ValidateTableRetention", new ValidateTableRetention());
diff --git
a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/PinotAdministrator.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/PinotAdministrator.java
index 8a2c8edcf39..e2848510650 100644
---
a/pinot-tools/src/main/java/org/apache/pinot/tools/admin/PinotAdministrator.java
+++
b/pinot-tools/src/main/java/org/apache/pinot/tools/admin/PinotAdministrator.java
@@ -23,6 +23,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.pinot.common.Utils;
import org.apache.pinot.common.utils.tls.JvmDefaultSslContext;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.tools.Command;
import org.apache.pinot.tools.admin.command.AddSchemaCommand;
@@ -94,6 +95,7 @@ public class PinotAdministrator {
private static final Map<String, Command> SUBCOMMAND_MAP = new HashMap<>();
static {
+ ZooKeeperSslContextSupplier.initialize();
JvmDefaultSslContext.initDefaultSslContext();
SUBCOMMAND_MAP.put("QuickStart", new QuickStartCommand());
SUBCOMMAND_MAP.put("OperateClusterConfig", new
OperateClusterConfigCommand());
diff --git
a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
index 3a500b25e9a..a4f12c0e6a3 100644
---
a/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
+++
b/pinot-tools/src/main/java/org/apache/pinot/tools/perf/PerfBenchmarkDriver.java
@@ -46,6 +46,7 @@ import
org.apache.helix.tools.ClusterVerifiers.StrictMatchExternalViewVerifier;
import org.apache.pinot.broker.broker.helix.HelixBrokerStarter;
import org.apache.pinot.common.auth.AuthProviderUtils;
import org.apache.pinot.common.utils.ZkStarter;
+import org.apache.pinot.common.utils.tls.ZooKeeperSslContextSupplier;
import org.apache.pinot.controller.ControllerConf;
import org.apache.pinot.controller.ControllerStarter;
import org.apache.pinot.controller.helix.core.PinotHelixResourceManager;
@@ -71,6 +72,10 @@ public class PerfBenchmarkDriver {
private static final long BROKER_TIMEOUT_MS = 60_000L;
private static final String BROKER_QUERY_PATH = "/query/sql";
+ static {
+ ZooKeeperSslContextSupplier.initialize();
+ }
+
private final PerfBenchmarkDriverConf _conf;
private final String _zkAddress;
private final String _clusterName;
diff --git a/pom.xml b/pom.xml
index f96881cdbdd..8f32dfdcff5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -218,7 +218,7 @@
<libthrift.version>0.25.0</libthrift.version>
<log4j.version>2.26.1</log4j.version>
<slf4j.version>2.0.20</slf4j.version>
- <netty.version>4.1.139.Final</netty.version>
+ <netty.version>4.2.19.Final</netty.version>
<reactor-netty.version>1.3.7</reactor-netty.version>
<reactor-core.version>3.8.7</reactor-core.version>
<reactivestreams.version>1.0.4</reactivestreams.version>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
