raghavyadav01 commented on code in PR #20021: URL: https://github.com/apache/pinot/pull/20021#discussion_r4505323862
########## pinot-core/src/main/java/org/apache/pinot/core/transport/RequestWriteTimeoutHandler.java: ########## @@ -0,0 +1,206 @@ +/** + * 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 { Review Comment: Could Netty's `IdleStateHandler` with `observeOutput=true` (writer-idle, only acted on while bytes are pending) cover this instead of a custom handler? It already tracks outbound-buffer progress. Is there a case it misses that needs the per-write progressive promises? ########## pinot-common/src/main/java/org/apache/pinot/common/config/NettyConfig.java: ########## @@ -18,13 +18,56 @@ */ package org.apache.pinot.common.config; +import com.google.common.base.Preconditions; import org.apache.pinot.spi.env.PinotConfiguration; /// Container object for netty configuration of pinot clients and servers (netty, grizzly, etc.) +/// +/// Every setting is read as `<namespace>.<suffix>`, where the namespace is `pinot.broker.netty` on the broker and +/// `pinot.server.netty` on the server. The channel liveness settings below only apply to the client side, i.e. the +/// broker-to-server channels opened by `ServerChannels`; the server-side `QueryServer` ignores them. public class NettyConfig { private static final String NATIVE_TRANSPORTS_ENABLED = "native.transports.enabled"; + + /// Maximum interval without positive socket byte progress while outbound writes are pending, in milliseconds. + /// Bytes accepted by the kernel refresh the interval for the whole connection, so a slow progressing request and + /// requests queued behind it can take longer overall. On a half-open or stalled socket the peer stops acknowledging, + /// the send buffer fills, and `isActive()` can remain `true` for many minutes. Closing the channel fails queries fast + /// and lets the failure detector quarantine the server. Non-positive disables the check. Broker key: + /// `pinot.broker.netty.write.timeout.ms`. + public static final String WRITE_TIMEOUT_MS = "write.timeout.ms"; + public static final long DEFAULT_WRITE_TIMEOUT_MS = 10_000L; + + /// Linux `TCP_USER_TIMEOUT`, in milliseconds: how long transmitted data may remain unacknowledged before the + /// kernel closes the connection, replacing the default retransmission timeout of several minutes. Covers the + /// case the write timeout cannot see: small writes that the kernel accepted into a send buffer that the peer never + /// drains. Only applied when the broker uses the Epoll transport (`native.transports.enabled` on Linux); NIO and + /// KQueue have no such socket option. Non-positive leaves the kernel default. Broker key: + /// `pinot.broker.netty.tcp.user.timeout.ms`. + public static final String TCP_USER_TIMEOUT_MS = "tcp.user.timeout.ms"; + public static final int DEFAULT_TCP_USER_TIMEOUT_MS = 10_000; + + /// Netty per-channel outbound queue high watermark, in bytes. Once the bytes queued for a channel exceed it, + /// `Channel.isWritable()` turns `false` and `ServerChannels` rejects further requests to that server instead of + /// queueing them, so a closed send window costs at most this much direct memory per channel rather than growing + /// without bound. Requests are small, and a healthy peer drains its socket at line rate, so the queue only reaches + /// this depth when the peer or the path has stalled. Broker key: + /// `pinot.broker.netty.write.buffer.high.water.mark.bytes`. + public static final String WRITE_BUFFER_HIGH_WATER_MARK_BYTES = "write.buffer.high.water.mark.bytes"; + public static final int DEFAULT_WRITE_BUFFER_HIGH_WATER_MARK_BYTES = 16 * 1024 * 1024; Review Comment: Since the instance request carries the segment list, a burst of queries to a server on a table with many segments could cross 16 MiB and fail queries that used to just queue. Should the rejection start disabled (or the default be sized from real request sizes) until we have numbers from a busy broker? -- 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]
