xiangfu0 commented on code in PR #19873:
URL: https://github.com/apache/pinot/pull/19873#discussion_r4278301124
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -2875,6 +2875,20 @@ public static class MultiStageQueryRunner {
"pinot.query.runner.mailbox.server.permit.keep.alive.without.calls";
public static final boolean
DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS = false;
+ /// Whether the mailbox server and the query server of an instance reject
a request addressed to another instance.
+ ///
+ /// Senders identify the instance they mean in the request headers. A
request reaches the wrong instance when the
+ /// sender's connection is stale: for example after two server pods
restart together and take each other's IP
+ /// addresses, while every server listens on the same port. If the wrong
instance accepts the request, the intended
+ /// one waits for data until the query times out, and the wrong one runs
the stage plan, or keeps the block in a
+ /// mailbox that nothing reads. When the wrong instance rejects the
request, the sender resets the connection and
+ /// the query fails with a clear error, in most cases at once rather than
at its deadline.
+ ///
+ /// Requests without the headers, as sent by older versions, are always
accepted. The value is read when the
+ /// instance starts, on brokers and servers.
+ public static final String KEY_OF_REJECT_MISROUTED_REQUESTS =
"pinot.query.runner.reject.misrouted.requests";
+ public static final boolean DEFAULT_REJECT_MISROUTED_REQUESTS = true;
Review Comment:
This is a new behavior-changing check defaulting to ON (C1.6 usually wants
new features OFF). It's backward-compatible since older senders are accepted,
but a true-by-default check that can *reject* traffic is exactly where a subtle
id mismatch (e.g. server's `CONFIG_OF_INSTANCE_ID` diverging from the Helix
instance name the broker routes by) would silently fail queries instead of
hanging them. Worth confirming in the description why ON-by-default is safe for
existing clusters, and whether a release-notes callout is warranted given it's
a default-on safety behavior.
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/routing/QueryServerInstance.java:
##########
@@ -60,6 +75,11 @@ public int getQueryServicePort() {
return _queryServicePort;
}
+ /// Returns the hash of the instance id. See [#hashInstanceId(String)].
+ public long getInstanceIdHash() {
+ return _instanceIdHash;
Review Comment:
Note that `equals`/`hashCode` (just below) deliberately exclude this hash
and the instance id, so two instances sharing hostname:port collapse to one key
— e.g. in `createMailboxInfo`'s `workersByUniqueHostPort` grouping. That's fine
at plan time (distinct instances have distinct hostnames), but now that the
identity hash rides along it's a latent trap if a future caller ever keys on
these objects expecting id-distinctness. A one-line comment noting the equality
contract is intentionally address-only would help.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -408,13 +453,83 @@ public void onCompleted() {
}
};
+ return openStream(responseObserver);
+ }
+
+ private ClientCallStreamObserver<MailboxContent>
openStream(StreamObserver<MailboxStatus> responseObserver) {
+ Metadata metadata = new Metadata();
+ metadata.put(ChannelUtils.MAILBOX_ID_METADATA_KEY, _id);
+ ReceiverIdentity.addTargetInstanceIdHash(metadata,
_receiverInstanceIdHash);
return (ClientCallStreamObserver<MailboxContent>) PinotMailboxGrpc.newStub(
_channelManager.getChannel(_hostname, _port))
.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata))
.withDeadlineAfter(_deadlineMs - System.currentTimeMillis(),
TimeUnit.MILLISECONDS)
.open(responseObserver);
}
+ /// Handles the rejection of the stream by an instance that is not the
receiver the plan meant. Runs on a gRPC
+ /// thread, after [#_misroutedError] is set.
+ ///
+ /// The channel to [#_hostname] leads to another instance, typically because
that instance took over the IP address
+ /// the hostname used to resolve to and listens on the same port. Whatever
this mailbox sent went nowhere. So:
+ ///
+ /// - the channel is refreshed, so that the next stream to this address
resolves the hostname again;
+ /// - every later data send fails, which fails this stage and, through its
error EOS, the query;
+ /// - the intended receiver is told through a new stream, so that it fails
now instead of waiting for data until the
+ /// query times out. This is best effort, see [#rerouteError].
Review Comment:
The best-effort reroute here is nicely documented, but the "DNS still
returns the stale address so the reroute stream is itself rejected" branch (the
`onError`/`isMisrouted` path in `rerouteError`) looks untested — every test
remaps DNS to the new A before the reroute runs, so it always succeeds. Since
that's the whole reason this path is best-effort, a test that leaves DNS
pointing at B would be worth adding to lock in the second `refreshChannel`
behavior.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java:
##########
@@ -295,6 +311,94 @@ public void
testAllLeafStagesEmptyBrokerResponsesWithPhysicalOptimizer()
List.of(), "LONG", "LONG");
}
+ /// The mailbox and query servers of a running cluster reject requests that
say they are meant for another instance.
+ /// Shows that the instance ids reach the receivers of a real server and
broker, and that the check is on by default.
+ /// The other tests of this class show that the requests of real plans,
which carry the ids, are accepted.
+ @Test
+ public void testReceiversRejectRequestsForAnotherInstance()
+ throws Exception {
+ String otherInstanceId = "Server_other-host_8098";
+ String serverInstanceId = _serverStarters.get(0).getInstanceId();
+ InstanceConfig serverConfig =
_helixAdmin.getInstanceConfig(getHelixClusterName(), serverInstanceId);
+ // Every instance of the test cluster runs on this host
+ String host = "localhost";
+
+ // Stage plan request (the cancel of an unknown query, which is harmless
when accepted)
+ int queryServerPort = serverConfig.getRecord()
+
.getIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_SERVICE_PORT_KEY,
-1);
+ ManagedChannel queryServerChannel = ManagedChannelBuilder.forAddress(host,
queryServerPort).usePlaintext().build();
+ try {
+ Metadata headers = new Metadata();
+ ReceiverIdentity.addTargetInstanceIdHash(headers,
QueryServerInstance.hashInstanceId(otherInstanceId));
+ StatusRuntimeException e = expectThrows(StatusRuntimeException.class, ()
-> cancel(queryServerChannel, headers));
+ assertMisroutedTo(e, serverInstanceId);
+
+ Metadata ownHeaders = new Metadata();
+ ReceiverIdentity.addTargetInstanceIdHash(ownHeaders,
QueryServerInstance.hashInstanceId(serverInstanceId));
+ cancel(queryServerChannel, ownHeaders);
+ } finally {
+ queryServerChannel.shutdownNow();
+ }
+
+ // Mailbox streams, to the server and to the broker
+ int serverMailboxPort = serverConfig.getRecord()
+
.getIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_MAILBOX_PORT_KEY,
-1);
+ assertMailboxStreamRejected(host, serverMailboxPort, otherInstanceId,
serverInstanceId);
+ String brokerInstanceId = _brokerStarters.get(0).getInstanceId();
+ InstanceConfig brokerConfig =
_helixAdmin.getInstanceConfig(getHelixClusterName(), brokerInstanceId);
+ int brokerMailboxPort = brokerConfig.getRecord()
+
.getIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_MAILBOX_PORT_KEY,
-1);
+ assertMailboxStreamRejected(host, brokerMailboxPort, otherInstanceId,
brokerInstanceId);
+ }
+
+ private static void cancel(ManagedChannel channel, Metadata headers) {
+
PinotQueryWorkerGrpc.newBlockingStub(channel).withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers))
+ .withDeadlineAfter(10,
TimeUnit.SECONDS).cancel(Worker.CancelRequest.newBuilder().setRequestId(-1).build());
+ }
+
+ private static void assertMailboxStreamRejected(String host, int port,
String targetInstanceId,
+ String receiverInstanceId)
+ throws Exception {
+ ManagedChannel channel = ManagedChannelBuilder.forAddress(host,
port).usePlaintext().build();
+ try {
+ Metadata headers = new Metadata();
+ headers.put(ChannelUtils.MAILBOX_ID_METADATA_KEY, "-1|1|0|0|0");
+ ReceiverIdentity.addTargetInstanceIdHash(headers,
QueryServerInstance.hashInstanceId(targetInstanceId));
+ CompletableFuture<Throwable> error = new CompletableFuture<>();
+ StreamObserver<Mailbox.MailboxContent> stream =
PinotMailboxGrpc.newStub(channel)
+ .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers))
+ .withDeadlineAfter(10, TimeUnit.SECONDS)
+ .open(new StreamObserver<>() {
+ @Override
+ public void onNext(Mailbox.MailboxStatus value) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ error.complete(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ error.complete(null);
+ }
+ });
+
stream.onNext(Mailbox.MailboxContent.newBuilder().setMailboxId("-1|1|0|0|0").setWaitForMore(true).build());
+ Throwable t = error.get(10, TimeUnit.SECONDS);
+ assertNotNull(t, "The mailbox server at " + host + ":" + port + "
accepted a stream for another instance");
+ assertMisroutedTo(t, receiverInstanceId);
+ } finally {
+ channel.shutdownNow();
+ }
+ }
+
+ /// Asserts that `t` is the rejection of a request by the instance with the
given id.
+ private static void assertMisroutedTo(Throwable t, String
receiverInstanceId) {
+ String misroutedTo = ReceiverIdentity.getMisroutedTo(t);
+ assertNotNull(misroutedTo, "Not a rejection of a request meant for another
instance: " + t);
+ assertEquals(misroutedTo.split(" ")[0], receiverInstanceId);
Review Comment:
Minor: `misroutedTo.split(" ")[0]` relies on the exact `"<id> at <addr>"`
shape of `ReceiverIdentity.toString()`. If that description format ever changes
this assertion breaks confusingly. A `startsWith(receiverInstanceId + " ")` (or
asserting it contains the id) would be a touch more robust.
##########
pinot-query-planner/src/main/java/org/apache/pinot/query/routing/QueryServerInstance.java:
##########
@@ -60,6 +75,11 @@ public int getQueryServicePort() {
return _queryServicePort;
}
+ /// Returns the hash of the instance id. See [#hashInstanceId(String)].
+ public long getInstanceIdHash() {
+ return _instanceIdHash;
Review Comment:
Note that `equals`/`hashCode` (just below) deliberately exclude this hash
and the instance id, so two instances sharing hostname:port collapse to one key
— e.g. in `createMailboxInfo`'s `workersByUniqueHostPort` grouping. That's fine
at plan time (distinct instances have distinct hostnames), but now that the
identity hash rides along it's a latent trap if a future caller ever keys on
these objects expecting id-distinctness. A one-line comment noting the equality
contract is intentionally address-only would help.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java:
##########
@@ -408,13 +453,83 @@ public void onCompleted() {
}
};
+ return openStream(responseObserver);
+ }
+
+ private ClientCallStreamObserver<MailboxContent>
openStream(StreamObserver<MailboxStatus> responseObserver) {
+ Metadata metadata = new Metadata();
+ metadata.put(ChannelUtils.MAILBOX_ID_METADATA_KEY, _id);
+ ReceiverIdentity.addTargetInstanceIdHash(metadata,
_receiverInstanceIdHash);
return (ClientCallStreamObserver<MailboxContent>) PinotMailboxGrpc.newStub(
_channelManager.getChannel(_hostname, _port))
.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata))
.withDeadlineAfter(_deadlineMs - System.currentTimeMillis(),
TimeUnit.MILLISECONDS)
.open(responseObserver);
}
+ /// Handles the rejection of the stream by an instance that is not the
receiver the plan meant. Runs on a gRPC
+ /// thread, after [#_misroutedError] is set.
+ ///
+ /// The channel to [#_hostname] leads to another instance, typically because
that instance took over the IP address
+ /// the hostname used to resolve to and listens on the same port. Whatever
this mailbox sent went nowhere. So:
+ ///
+ /// - the channel is refreshed, so that the next stream to this address
resolves the hostname again;
+ /// - every later data send fails, which fails this stage and, through its
error EOS, the query;
+ /// - the intended receiver is told through a new stream, so that it fails
now instead of waiting for data until the
+ /// query times out. This is best effort, see [#rerouteError].
Review Comment:
The best-effort reroute here is nicely documented, but the "DNS still
returns the stale address so the reroute stream is itself rejected" branch (the
`onError`/`isMisrouted` path in `rerouteError`) looks untested — every test
remaps DNS to the new A before the reroute runs, so it always succeeds. Since
that's the whole reason this path is best-effort, a test that leaves DNS
pointing at B would be worth adding to lock in the second `refreshChannel`
behavior.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -2875,6 +2875,20 @@ public static class MultiStageQueryRunner {
"pinot.query.runner.mailbox.server.permit.keep.alive.without.calls";
public static final boolean
DEFAULT_OF_MAILBOX_SERVER_PERMIT_KEEP_ALIVE_WITHOUT_CALLS = false;
+ /// Whether the mailbox server and the query server of an instance reject
a request addressed to another instance.
+ ///
+ /// Senders identify the instance they mean in the request headers. A
request reaches the wrong instance when the
+ /// sender's connection is stale: for example after two server pods
restart together and take each other's IP
+ /// addresses, while every server listens on the same port. If the wrong
instance accepts the request, the intended
+ /// one waits for data until the query times out, and the wrong one runs
the stage plan, or keeps the block in a
+ /// mailbox that nothing reads. When the wrong instance rejects the
request, the sender resets the connection and
+ /// the query fails with a clear error, in most cases at once rather than
at its deadline.
+ ///
+ /// Requests without the headers, as sent by older versions, are always
accepted. The value is read when the
+ /// instance starts, on brokers and servers.
+ public static final String KEY_OF_REJECT_MISROUTED_REQUESTS =
"pinot.query.runner.reject.misrouted.requests";
+ public static final boolean DEFAULT_REJECT_MISROUTED_REQUESTS = true;
Review Comment:
This is a new behavior-changing check defaulting to ON (C1.6 usually wants
new features OFF). It's backward-compatible since older senders are accepted,
but a true-by-default check that can *reject* traffic is exactly where a subtle
id mismatch (e.g. server's `CONFIG_OF_INSTANCE_ID` diverging from the Helix
instance name the broker routes by) would silently fail queries instead of
hanging them. Worth confirming in the description why ON-by-default is safe for
existing clusters, and whether a release-notes callout is warranted given it's
a default-on safety behavior.
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MultiStageEngineIntegrationTest.java:
##########
@@ -295,6 +311,94 @@ public void
testAllLeafStagesEmptyBrokerResponsesWithPhysicalOptimizer()
List.of(), "LONG", "LONG");
}
+ /// The mailbox and query servers of a running cluster reject requests that
say they are meant for another instance.
+ /// Shows that the instance ids reach the receivers of a real server and
broker, and that the check is on by default.
+ /// The other tests of this class show that the requests of real plans,
which carry the ids, are accepted.
+ @Test
+ public void testReceiversRejectRequestsForAnotherInstance()
+ throws Exception {
+ String otherInstanceId = "Server_other-host_8098";
+ String serverInstanceId = _serverStarters.get(0).getInstanceId();
+ InstanceConfig serverConfig =
_helixAdmin.getInstanceConfig(getHelixClusterName(), serverInstanceId);
+ // Every instance of the test cluster runs on this host
+ String host = "localhost";
+
+ // Stage plan request (the cancel of an unknown query, which is harmless
when accepted)
+ int queryServerPort = serverConfig.getRecord()
+
.getIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_SERVICE_PORT_KEY,
-1);
+ ManagedChannel queryServerChannel = ManagedChannelBuilder.forAddress(host,
queryServerPort).usePlaintext().build();
+ try {
+ Metadata headers = new Metadata();
+ ReceiverIdentity.addTargetInstanceIdHash(headers,
QueryServerInstance.hashInstanceId(otherInstanceId));
+ StatusRuntimeException e = expectThrows(StatusRuntimeException.class, ()
-> cancel(queryServerChannel, headers));
+ assertMisroutedTo(e, serverInstanceId);
+
+ Metadata ownHeaders = new Metadata();
+ ReceiverIdentity.addTargetInstanceIdHash(ownHeaders,
QueryServerInstance.hashInstanceId(serverInstanceId));
+ cancel(queryServerChannel, ownHeaders);
+ } finally {
+ queryServerChannel.shutdownNow();
+ }
+
+ // Mailbox streams, to the server and to the broker
+ int serverMailboxPort = serverConfig.getRecord()
+
.getIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_MAILBOX_PORT_KEY,
-1);
+ assertMailboxStreamRejected(host, serverMailboxPort, otherInstanceId,
serverInstanceId);
+ String brokerInstanceId = _brokerStarters.get(0).getInstanceId();
+ InstanceConfig brokerConfig =
_helixAdmin.getInstanceConfig(getHelixClusterName(), brokerInstanceId);
+ int brokerMailboxPort = brokerConfig.getRecord()
+
.getIntField(CommonConstants.Helix.Instance.MULTI_STAGE_QUERY_ENGINE_MAILBOX_PORT_KEY,
-1);
+ assertMailboxStreamRejected(host, brokerMailboxPort, otherInstanceId,
brokerInstanceId);
+ }
+
+ private static void cancel(ManagedChannel channel, Metadata headers) {
+
PinotQueryWorkerGrpc.newBlockingStub(channel).withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers))
+ .withDeadlineAfter(10,
TimeUnit.SECONDS).cancel(Worker.CancelRequest.newBuilder().setRequestId(-1).build());
+ }
+
+ private static void assertMailboxStreamRejected(String host, int port,
String targetInstanceId,
+ String receiverInstanceId)
+ throws Exception {
+ ManagedChannel channel = ManagedChannelBuilder.forAddress(host,
port).usePlaintext().build();
+ try {
+ Metadata headers = new Metadata();
+ headers.put(ChannelUtils.MAILBOX_ID_METADATA_KEY, "-1|1|0|0|0");
+ ReceiverIdentity.addTargetInstanceIdHash(headers,
QueryServerInstance.hashInstanceId(targetInstanceId));
+ CompletableFuture<Throwable> error = new CompletableFuture<>();
+ StreamObserver<Mailbox.MailboxContent> stream =
PinotMailboxGrpc.newStub(channel)
+ .withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers))
+ .withDeadlineAfter(10, TimeUnit.SECONDS)
+ .open(new StreamObserver<>() {
+ @Override
+ public void onNext(Mailbox.MailboxStatus value) {
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ error.complete(t);
+ }
+
+ @Override
+ public void onCompleted() {
+ error.complete(null);
+ }
+ });
+
stream.onNext(Mailbox.MailboxContent.newBuilder().setMailboxId("-1|1|0|0|0").setWaitForMore(true).build());
+ Throwable t = error.get(10, TimeUnit.SECONDS);
+ assertNotNull(t, "The mailbox server at " + host + ":" + port + "
accepted a stream for another instance");
+ assertMisroutedTo(t, receiverInstanceId);
+ } finally {
+ channel.shutdownNow();
+ }
+ }
+
+ /// Asserts that `t` is the rejection of a request by the instance with the
given id.
+ private static void assertMisroutedTo(Throwable t, String
receiverInstanceId) {
+ String misroutedTo = ReceiverIdentity.getMisroutedTo(t);
+ assertNotNull(misroutedTo, "Not a rejection of a request meant for another
instance: " + t);
+ assertEquals(misroutedTo.split(" ")[0], receiverInstanceId);
Review Comment:
Minor: `misroutedTo.split(" ")[0]` relies on the exact `"<id> at <addr>"`
shape of `ReceiverIdentity.toString()`. If that description format ever changes
this assertion breaks confusingly. A `startsWith(receiverInstanceId + " ")` (or
asserting it contains the id) would be a touch more robust.
--
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]
