This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 8984d8c589e Include stage/worker/server provenance in MSE error block
logs (#19002)
8984d8c589e is described below
commit 8984d8c589e2b0010528fc4a977429364f35f5f9
Author: Gonzalo Ortiz Jaureguizar <[email protected]>
AuthorDate: Thu Jul 16 20:14:00 2026 +0200
Include stage/worker/server provenance in MSE error block logs (#19002)
---
.../apache/pinot/query/runtime/QueryRunner.java | 4 +-
.../pinot/query/runtime/blocks/ErrorMseBlock.java | 6 +++
.../runtime/executor/OpChainSchedulerService.java | 5 +-
.../query/runtime/blocks/ErrorMseBlockTest.java | 62 ++++++++++++++++++++++
4 files changed, 74 insertions(+), 3 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
index 910f53480e4..e5575d7680f 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java
@@ -368,8 +368,8 @@ public class QueryRunner {
StageMetadata stageMetadata = stagePlan.getStageMetadata();
QueryExecutionContext queryContext =
QueryThreadContext.get().getExecutionContext();
long requestId = queryContext.getRequestId();
- LOGGER.error("Error executing stage for request: {}, stage: {}, sending
error block: {}", requestId,
- stageMetadata.getStageId(), errorBlock);
+ // The stage/worker/server are already rendered by errorBlock.toString();
only requestId is added here.
+ LOGGER.error("Error executing stage for request: {}, sending error block:
{}", requestId, errorBlock);
try {
OpChainExecutionContext executionContext =
OpChainExecutionContext.fromQueryContext(_mailboxService,
opChainMetadata, stageMetadata, workerMetadata,
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlock.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlock.java
index 6e02536ecef..11a8fb2d67f 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlock.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlock.java
@@ -131,6 +131,12 @@ public class ErrorMseBlock implements MseBlock.Eos {
try {
ObjectNode root = JsonUtils.newObjectNode();
root.put("type", "error");
+ // Provenance of the error: the stage, worker and server where the error
originated. This is mostly useful when
+ // the block reaches downstream logs (e.g. the broker or an intermediate
stage), so operators can quickly find
+ // where the failure actually happened. -1 / null means the origin could
not be determined (see #fromMap).
+ root.put("stageId", _stageId);
+ root.put("workerId", _workerId);
+ root.put("serverId", _serverId);
root.set("errorMessages", JsonUtils.objectToJsonNode(_errorMessages));
return JsonUtils.objectToString(root);
} catch (JsonProcessingException e) {
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
index 75de3ad8eda..41719090853 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/executor/OpChainSchedulerService.java
@@ -207,8 +207,11 @@ public class OpChainSchedulerService {
statsRef.set(stats);
if (result.isError()) {
ErrorMseBlock errorBlock = (ErrorMseBlock) result;
+ String serverId = errorBlock.getServerId() != null ?
errorBlock.getServerId() : "unknown";
throw errorBlock.getMainErrorCode().asException("Error block "
- + "from " + errorBlock.getServerId()
+ + "from stage " + errorBlock.getStageId()
+ + " worker " + errorBlock.getWorkerId()
+ + " on " + serverId
+ ". Msg: " + errorBlock.getErrorMessages()
+ ". Stats: " + stats);
} else {
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlockTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlockTest.java
new file mode 100644
index 00000000000..361f3f70328
--- /dev/null
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/blocks/ErrorMseBlockTest.java
@@ -0,0 +1,62 @@
+/**
+ * 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.query.runtime.blocks;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.util.Map;
+import org.apache.pinot.spi.exception.QueryErrorCode;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+public class ErrorMseBlockTest {
+
+ @Test
+ public void toStringIncludesProvenance()
+ throws Exception {
+ ErrorMseBlock block =
+ new ErrorMseBlock(3, 7, "Server_localhost_8000",
Map.of(QueryErrorCode.INTERNAL, "boom"));
+
+ JsonNode json = JsonUtils.stringToJsonNode(block.toString());
+
+ assertEquals(json.get("type").asText(), "error");
+ // stage/worker/server are surfaced so downstream logs can locate where
the failure originated.
+ assertEquals(json.get("stageId").asInt(), 3);
+ assertEquals(json.get("workerId").asInt(), 7);
+ assertEquals(json.get("serverId").asText(), "Server_localhost_8000");
+ assertTrue(json.get("errorMessages").toString().contains("boom"));
+ }
+
+ @Test
+ public void toStringHandlesUnknownProvenance()
+ throws Exception {
+ // The deprecated constructor leaves the origin undetermined:
stageId/workerId = -1, serverId = null.
+ @SuppressWarnings("deprecation")
+ ErrorMseBlock block = new ErrorMseBlock(Map.of(QueryErrorCode.UNKNOWN, "no
context"));
+
+ JsonNode json = JsonUtils.stringToJsonNode(block.toString());
+
+ assertEquals(json.get("stageId").asInt(), -1);
+ assertEquals(json.get("workerId").asInt(), -1);
+ assertTrue(json.get("serverId").isNull(), "serverId should serialize as
null when the origin is unknown");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
