xiangfu0 commented on code in PR #20021:
URL: https://github.com/apache/pinot/pull/20021#discussion_r4369034187
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -365,11 +486,24 @@ private void awaitTlsHandshake(Channel channel, long
timeoutMs)
}
}
- void sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
+ /// Returns the channel the request was written to.
+ Channel sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, byte[] requestBytes) {
- long startTimeMs = System.currentTimeMillis();
- // The listener runs later, by when a reconnect may have replaced
_channel: on failure it closes this connection.
+ // Pin the channel: the completion listener must act on the channel it
wrote to, not on whatever channel this
+ // entry holds by the time the write completes, or a late failure would
close a healthy reconnected channel.
Channel channel = _channel;
+ if (!channel.isWritable()) {
Review Comment:
Behavior note: for non-skip queries an unwritable channel now fails the
whole query (vs. previously queueing in direct memory). That's the intended
trade-off and the default 16MB high watermark means it only triggers on a
genuine stall, but it is a user-visible behavior change worth a line in the PR
description / release notes.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/RequestWriteTimeoutHandler.java:
##########
@@ -0,0 +1,205 @@
+/**
+ * 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.core.transport;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundBuffer;
+import io.netty.channel.ChannelProgressiveFuture;
+import io.netty.channel.ChannelProgressiveFutureListener;
+import io.netty.channel.ChannelProgressivePromise;
+import io.netty.channel.ChannelPromise;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.BrokerMeter;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Broker-side socket-idle monitor. While outbound bytes are pending, closes
a server channel if the kernel accepts
+/// no bytes for the configured timeout. Progress by the current queue head
keeps later queued writes alive; enqueue,
+/// cancellation and write failure do not reset the idle clock.
+///
+/// Install on the socket side of SSL so ciphertext progress is observable.
This does not time plaintext waiting
+/// inside SSL before ciphertext exists; SSL's handshake timeout and the
pre-connect handshake budget remain separate.
+/// Each socket write uses a progressive promise, relaying the downstream
outcome to the original promise without
+/// completing it early or releasing Netty-owned buffers. One connection timer
runs only while writes are pending.
+///
+/// Not sharable: all state is confined to the channel's event loop.
+public class RequestWriteTimeoutHandler extends ChannelDuplexHandler {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(RequestWriteTimeoutHandler.class);
+
+ private final ServerRoutingInstance _serverRoutingInstance;
+ private final long _timeoutMs;
+ private final long _timeoutNanos;
+ private final Set<PendingWrite> _pendingWrites = new HashSet<>();
+ private Set<PendingWrite> _unflushedWrites = new HashSet<>();
+ private long _lastProgressNanos;
+ private ScheduledFuture<?> _timer;
+ private boolean _closed;
+
+ public RequestWriteTimeoutHandler(ServerRoutingInstance
serverRoutingInstance, long timeoutMs) {
+ _serverRoutingInstance = serverRoutingInstance;
+ _timeoutMs = timeoutMs;
+ _timeoutNanos = TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ }
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) {
+ // SSL supplies regular promises for ciphertext. Void writes also need
monitoring; unvoid preserves their
+ // exception propagation while giving the relay a real completion signal.
+ PendingWrite write = new PendingWrite(ctx, promise.unvoid(), msg
instanceof ByteBuf
+ && ((ByteBuf) msg).isReadable());
+ ctx.write(msg, write._progressivePromise);
+ }
+
+ @Override
+ public void flush(ChannelHandlerContext ctx) {
+ // Netty makes flushed outbound promises uncancelable. Preserve that
contract for the caller's promise too,
+ // while cancellation before flush still cancels the queued downstream
promise so Netty releases/skips it.
+ if (!_unflushedWrites.isEmpty()) {
+ Set<PendingWrite> unflushedWrites = _unflushedWrites;
+ // Detach this first-flush batch before callbacks can complete writes or
remove the handler reentrantly.
+ _unflushedWrites = new HashSet<>();
+ for (PendingWrite write : unflushedWrites) {
+ if (_pendingWrites.contains(write) &&
!write._promise.setUncancellable()) {
+ write._progressivePromise.cancel(false);
+ }
+ }
+ }
+ ctx.flush();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) {
+ stopMonitoring();
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) {
+ stopMonitoring();
+ }
+
+ private void stopMonitoring() {
+ _closed = true;
+ cancelTimer();
+ // The listeners still relay real downstream outcomes after removal; do
not complete or release anything here.
+ _pendingWrites.clear();
+ _unflushedWrites.clear();
+ }
+
+ private void cancelTimer() {
+ if (_timer != null) {
+ _timer.cancel(false);
+ _timer = null;
+ }
+ }
+
+ private void scheduleTimeout(ChannelHandlerContext ctx, long delayNanos) {
+ _timer = ctx.executor().schedule(() -> checkTimeout(ctx), delayNanos,
TimeUnit.NANOSECONDS);
+ }
+
+ private void checkTimeout(ChannelHandlerContext ctx) {
+ _timer = null;
+ if (_closed || _pendingWrites.isEmpty()) {
+ return;
+ }
+ long idleNanos = System.nanoTime() - _lastProgressNanos;
+ if (idleNanos < _timeoutNanos) {
+ scheduleTimeout(ctx, _timeoutNanos - idleNanos);
+ return;
+ }
+ _closed = true;
+ Channel channel = ctx.channel();
+ ChannelOutboundBuffer outboundBuffer = channel.unsafe().outboundBuffer();
+ long pendingBytes = outboundBuffer != null ?
outboundBuffer.totalPendingWriteBytes() : -1;
+ LOGGER.error("Request write to server: {} made no progress for {}ms ({}
bytes pending, writable: {}), closing the "
+ + "channel as the socket is stalled or half-open",
_serverRoutingInstance, _timeoutMs, pendingBytes,
+ channel.isWritable());
+
BrokerMetrics.get().addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_WRITE_TIMEOUTS,
1);
+ ctx.close();
+ }
+
+ /// Per-write outcome relay; progress updates the shared socket clock, never
a separate write deadline.
+ private final class PendingWrite implements ChannelProgressiveFutureListener
{
+ final ChannelPromise _promise;
+ final ChannelProgressivePromise _progressivePromise;
+ final boolean _hasBytes;
+ long _progress;
+
+ PendingWrite(ChannelHandlerContext ctx, ChannelPromise promise, boolean
hasBytes) {
+ _promise = promise;
+ _hasBytes = hasBytes;
+ _progressivePromise = ctx.newProgressivePromise();
+ _pendingWrites.add(this);
+ _unflushedWrites.add(this);
+ if (_pendingWrites.size() == 1 && !_closed) {
Review Comment:
Nice: the idle clock is shared per-connection and only (re)armed when the
pending set goes 0->1, so queued writes behind the head can't indefinitely
postpone the timeout. The `testLaterEnqueuesDoNotPostponeSocketIdleTimeout`
test pins this well.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -447,11 +478,16 @@ private void processInstanceConfigChangeInternal() {
// Calculate new enabled and disabled servers
Set<String> enabledServers = new HashSet<>();
+ // Every server that still has an instance config, enabled or not
+ Set<String> presentServers = new HashSet<>();
List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
// Put instance initialization logics into try-catch block to prevent
bad server configs affecting the entire
// cluster
String instanceId = instanceConfigZNRecord.getId();
+ if (InstanceTypeUtils.isServer(instanceId)) {
+ presentServers.add(instanceId);
Review Comment:
Minor: `presentServers` is keyed by the raw `instanceConfigZNRecord.getId()`
here, while `_disabledServers`/`_enabledServerInstanceMap` are keyed by the
interned id. The later `presentServers.contains(...)` checks still work because
HashSet uses `.equals`, but it relies on raw and interned ids being
value-equal. Not a bug, just worth a one-line note so a future reader doesn't
assume identity matching.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/ServerChannels.java:
##########
@@ -365,11 +486,24 @@ private void awaitTlsHandshake(Channel channel, long
timeoutMs)
}
}
- void sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
+ /// Returns the channel the request was written to.
+ Channel sendRequestWithoutLocking(String rawTableName, AsyncQueryResponse
asyncQueryResponse,
ServerRoutingInstance serverRoutingInstance, byte[] requestBytes) {
- long startTimeMs = System.currentTimeMillis();
- // The listener runs later, by when a reconnect may have replaced
_channel: on failure it closes this connection.
+ // Pin the channel: the completion listener must act on the channel it
wrote to, not on whatever channel this
+ // entry holds by the time the write completes, or a late failure would
close a healthy reconnected channel.
Channel channel = _channel;
+ if (!channel.isWritable()) {
Review Comment:
Behavior note: for non-skip queries an unwritable channel now fails the
whole query (vs. previously queueing in direct memory). That's the intended
trade-off and the default 16MB high watermark means it only triggers on a
genuine stall, but it is a user-visible behavior change worth a line in the PR
description / release notes.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/routing/manager/BaseBrokerRoutingManager.java:
##########
@@ -447,11 +478,16 @@ private void processInstanceConfigChangeInternal() {
// Calculate new enabled and disabled servers
Set<String> enabledServers = new HashSet<>();
+ // Every server that still has an instance config, enabled or not
+ Set<String> presentServers = new HashSet<>();
List<String> newEnabledServers = new ArrayList<>();
for (ZNRecord instanceConfigZNRecord : instanceConfigZNRecords) {
// Put instance initialization logics into try-catch block to prevent
bad server configs affecting the entire
// cluster
String instanceId = instanceConfigZNRecord.getId();
+ if (InstanceTypeUtils.isServer(instanceId)) {
+ presentServers.add(instanceId);
Review Comment:
Minor: `presentServers` is keyed by the raw `instanceConfigZNRecord.getId()`
here, while `_disabledServers`/`_enabledServerInstanceMap` are keyed by the
interned id. The later `presentServers.contains(...)` checks still work because
HashSet uses `.equals`, but it relies on raw and interned ids being
value-equal. Not a bug, just worth a one-line note so a future reader doesn't
assume identity matching.
##########
pinot-core/src/main/java/org/apache/pinot/core/transport/RequestWriteTimeoutHandler.java:
##########
@@ -0,0 +1,205 @@
+/**
+ * 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.core.transport;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundBuffer;
+import io.netty.channel.ChannelProgressiveFuture;
+import io.netty.channel.ChannelProgressiveFutureListener;
+import io.netty.channel.ChannelProgressivePromise;
+import io.netty.channel.ChannelPromise;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.metrics.BrokerMeter;
+import org.apache.pinot.common.metrics.BrokerMetrics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Broker-side socket-idle monitor. While outbound bytes are pending, closes
a server channel if the kernel accepts
+/// no bytes for the configured timeout. Progress by the current queue head
keeps later queued writes alive; enqueue,
+/// cancellation and write failure do not reset the idle clock.
+///
+/// Install on the socket side of SSL so ciphertext progress is observable.
This does not time plaintext waiting
+/// inside SSL before ciphertext exists; SSL's handshake timeout and the
pre-connect handshake budget remain separate.
+/// Each socket write uses a progressive promise, relaying the downstream
outcome to the original promise without
+/// completing it early or releasing Netty-owned buffers. One connection timer
runs only while writes are pending.
+///
+/// Not sharable: all state is confined to the channel's event loop.
+public class RequestWriteTimeoutHandler extends ChannelDuplexHandler {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(RequestWriteTimeoutHandler.class);
+
+ private final ServerRoutingInstance _serverRoutingInstance;
+ private final long _timeoutMs;
+ private final long _timeoutNanos;
+ private final Set<PendingWrite> _pendingWrites = new HashSet<>();
+ private Set<PendingWrite> _unflushedWrites = new HashSet<>();
+ private long _lastProgressNanos;
+ private ScheduledFuture<?> _timer;
+ private boolean _closed;
+
+ public RequestWriteTimeoutHandler(ServerRoutingInstance
serverRoutingInstance, long timeoutMs) {
+ _serverRoutingInstance = serverRoutingInstance;
+ _timeoutMs = timeoutMs;
+ _timeoutNanos = TimeUnit.MILLISECONDS.toNanos(timeoutMs);
+ }
+
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise
promise) {
+ // SSL supplies regular promises for ciphertext. Void writes also need
monitoring; unvoid preserves their
+ // exception propagation while giving the relay a real completion signal.
+ PendingWrite write = new PendingWrite(ctx, promise.unvoid(), msg
instanceof ByteBuf
+ && ((ByteBuf) msg).isReadable());
+ ctx.write(msg, write._progressivePromise);
+ }
+
+ @Override
+ public void flush(ChannelHandlerContext ctx) {
+ // Netty makes flushed outbound promises uncancelable. Preserve that
contract for the caller's promise too,
+ // while cancellation before flush still cancels the queued downstream
promise so Netty releases/skips it.
+ if (!_unflushedWrites.isEmpty()) {
+ Set<PendingWrite> unflushedWrites = _unflushedWrites;
+ // Detach this first-flush batch before callbacks can complete writes or
remove the handler reentrantly.
+ _unflushedWrites = new HashSet<>();
+ for (PendingWrite write : unflushedWrites) {
+ if (_pendingWrites.contains(write) &&
!write._promise.setUncancellable()) {
+ write._progressivePromise.cancel(false);
+ }
+ }
+ }
+ ctx.flush();
+ }
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) {
+ stopMonitoring();
+ ctx.fireChannelInactive();
+ }
+
+ @Override
+ public void handlerRemoved(ChannelHandlerContext ctx) {
+ stopMonitoring();
+ }
+
+ private void stopMonitoring() {
+ _closed = true;
+ cancelTimer();
+ // The listeners still relay real downstream outcomes after removal; do
not complete or release anything here.
+ _pendingWrites.clear();
+ _unflushedWrites.clear();
+ }
+
+ private void cancelTimer() {
+ if (_timer != null) {
+ _timer.cancel(false);
+ _timer = null;
+ }
+ }
+
+ private void scheduleTimeout(ChannelHandlerContext ctx, long delayNanos) {
+ _timer = ctx.executor().schedule(() -> checkTimeout(ctx), delayNanos,
TimeUnit.NANOSECONDS);
+ }
+
+ private void checkTimeout(ChannelHandlerContext ctx) {
+ _timer = null;
+ if (_closed || _pendingWrites.isEmpty()) {
+ return;
+ }
+ long idleNanos = System.nanoTime() - _lastProgressNanos;
+ if (idleNanos < _timeoutNanos) {
+ scheduleTimeout(ctx, _timeoutNanos - idleNanos);
+ return;
+ }
+ _closed = true;
+ Channel channel = ctx.channel();
+ ChannelOutboundBuffer outboundBuffer = channel.unsafe().outboundBuffer();
+ long pendingBytes = outboundBuffer != null ?
outboundBuffer.totalPendingWriteBytes() : -1;
+ LOGGER.error("Request write to server: {} made no progress for {}ms ({}
bytes pending, writable: {}), closing the "
+ + "channel as the socket is stalled or half-open",
_serverRoutingInstance, _timeoutMs, pendingBytes,
+ channel.isWritable());
+
BrokerMetrics.get().addMeteredGlobalValue(BrokerMeter.NETTY_CONNECTION_WRITE_TIMEOUTS,
1);
+ ctx.close();
+ }
+
+ /// Per-write outcome relay; progress updates the shared socket clock, never
a separate write deadline.
+ private final class PendingWrite implements ChannelProgressiveFutureListener
{
+ final ChannelPromise _promise;
+ final ChannelProgressivePromise _progressivePromise;
+ final boolean _hasBytes;
+ long _progress;
+
+ PendingWrite(ChannelHandlerContext ctx, ChannelPromise promise, boolean
hasBytes) {
+ _promise = promise;
+ _hasBytes = hasBytes;
+ _progressivePromise = ctx.newProgressivePromise();
+ _pendingWrites.add(this);
+ _unflushedWrites.add(this);
+ if (_pendingWrites.size() == 1 && !_closed) {
Review Comment:
Nice: the idle clock is shared per-connection and only (re)armed when the
pending set goes 0->1, so queued writes behind the head can't indefinitely
postpone the timeout. The `testLaterEnqueuesDoNotPostponeSocketIdleTimeout`
test pins this well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
