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 c5e6b52c975 [pinot-server] Add consuming-segment decoder observability
(#19529)
c5e6b52c975 is described below
commit c5e6b52c975a66f07fb49c5993b9350f436907d6
Author: Anurag Rai <[email protected]>
AuthorDate: Fri Nov 13 09:07:03 2026 +0530
[pinot-server] Add consuming-segment decoder observability (#19529)
---
.../restlet/resources/SegmentConsumerInfo.java | 23 +++++++-
.../util/ConsumingSegmentInfoReader.java | 13 ++++-
.../ConsumingSegmentInfoReaderStatelessTest.java | 68 +++++++++++++++++++---
.../helix/RealtimeConsumerMonitorTest.java | 2 +-
.../realtime/RealtimeSegmentDataManager.java | 10 ++++
.../realtime/RealtimeSegmentDataManagerTest.java | 9 +++
.../pinot/server/api/resources/DebugResource.java | 3 +-
.../pinot/server/api/resources/TablesResource.java | 3 +-
8 files changed, 115 insertions(+), 16 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java
b/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java
index 9e695dd928e..c4f6d03f936 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentConsumerInfo.java
@@ -18,9 +18,11 @@
*/
package org.apache.pinot.common.restlet.resources;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.Map;
+import javax.annotation.Nullable;
/// Information regarding the consumer of a segment
@@ -31,17 +33,31 @@ public class SegmentConsumerInfo {
private final long _lastConsumedTimestamp;
private final Map<String, String> _partitionToOffsetMap;
private final PartitionOffsetInfo _partitionOffsetInfo;
+ // FQCN of the StreamMessageDecoder instantiated on the consumer. Null only
when the response comes from an older
+ // server that does not report it (e.g. during a rolling upgrade).
+ @Nullable
+ private final String _decoderClassName;
+
+ /// Kept for source and binary compatibility with callers that build this
DTO without the decoder class name. Not a
+ /// Jackson creator; deserialization goes through the `@JsonCreator`
constructor below.
+ public SegmentConsumerInfo(String segmentName, String consumerState, long
lastConsumedTimestamp,
+ Map<String, String> partitionToOffsetMap, PartitionOffsetInfo
partitionOffsetInfo) {
+ this(segmentName, consumerState, lastConsumedTimestamp,
partitionToOffsetMap, partitionOffsetInfo, null);
+ }
+ @JsonCreator
public SegmentConsumerInfo(@JsonProperty("segmentName") String segmentName,
@JsonProperty("consumerState") String consumerState,
@JsonProperty("lastConsumedTimestamp") long lastConsumedTimestamp,
@JsonProperty("partitionToOffsetMap") Map<String, String>
partitionToOffsetMap,
- @JsonProperty("partitionOffsetInfo") PartitionOffsetInfo
partitionOffsetInfo) {
+ @JsonProperty("partitionOffsetInfo") PartitionOffsetInfo
partitionOffsetInfo,
+ @JsonProperty("decoderClassName") @Nullable String decoderClassName) {
_segmentName = segmentName;
_consumerState = consumerState;
_lastConsumedTimestamp = lastConsumedTimestamp;
_partitionToOffsetMap = partitionToOffsetMap;
_partitionOffsetInfo = partitionOffsetInfo;
+ _decoderClassName = decoderClassName;
}
public String getSegmentName() {
@@ -64,6 +80,11 @@ public class SegmentConsumerInfo {
return _partitionOffsetInfo;
}
+ @Nullable
+ public String getDecoderClassName() {
+ return _decoderClassName;
+ }
+
@JsonIgnoreProperties(ignoreUnknown = true)
static public class PartitionOffsetInfo {
@JsonProperty("currentOffsets")
diff --git
a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
index 2cd851fe348..0fa2e14c1f9 100644
---
a/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
+++
b/pinot-controller/src/main/java/org/apache/pinot/controller/util/ConsumingSegmentInfoReader.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.controller.util;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.type.TypeReference;
@@ -31,6 +32,7 @@ import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;
+import javax.annotation.Nullable;
import org.apache.hc.client5.http.io.HttpClientConnectionManager;
import org.apache.pinot.common.exception.InvalidConfigException;
import org.apache.pinot.common.restlet.resources.SegmentConsumerInfo;
@@ -83,7 +85,7 @@ public class ConsumingSegmentInfoReader {
partitionOffsetInfo.getAvailabilityLagMs());
consumingSegmentInfoMap.computeIfAbsent(info.getSegmentName(), k ->
new ArrayList<>()).add(
new ConsumingSegmentInfo(serverName, info.getConsumerState(),
info.getLastConsumedTimestamp(),
- partitionOffsetInfo.getCurrentOffsets(), offsetInfo));
+ partitionOffsetInfo.getCurrentOffsets(), offsetInfo,
info.getDecoderClassName()));
}
}
// Segments which are in CONSUMING state but found no consumer on the
server
@@ -210,18 +212,23 @@ public class ConsumingSegmentInfoReader {
public Map<String, String> _partitionToOffsetMap;
@JsonProperty("partitionOffsetInfo")
public PartitionOffsetInfo _partitionOffsetInfo;
+ @JsonProperty("decoderClassName")
+ @Nullable
+ public String _decoderClassName;
-
+ @JsonCreator
public ConsumingSegmentInfo(@JsonProperty("serverName") String serverName,
@JsonProperty("consumerState") String consumerState,
@JsonProperty("lastConsumedTimestamp") long lastConsumedTimestamp,
@JsonProperty("partitionToOffsetMap") Map<String, String>
partitionToOffsetMap,
- @JsonProperty("partitionOffsetInfo") PartitionOffsetInfo
partitionOffsetInfo) {
+ @JsonProperty("partitionOffsetInfo") PartitionOffsetInfo
partitionOffsetInfo,
+ @JsonProperty("decoderClassName") @Nullable String decoderClassName) {
_serverName = serverName;
_consumerState = consumerState;
_lastConsumedTimestamp = lastConsumedTimestamp;
_partitionToOffsetMap = partitionToOffsetMap;
_partitionOffsetInfo = partitionOffsetInfo;
+ _decoderClassName = decoderClassName;
}
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/ConsumingSegmentInfoReaderStatelessTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/ConsumingSegmentInfoReaderStatelessTest.java
index 20d438ac9cc..6eb558fda73 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/api/ConsumingSegmentInfoReaderStatelessTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/api/ConsumingSegmentInfoReaderStatelessTest.java
@@ -67,6 +67,9 @@ public class ConsumingSegmentInfoReaderStatelessTest {
private static final String TABLE_NAME = "myTable_REALTIME";
private static final String SEGMENT_NAME_PARTITION_0 = "table__0__29__12345";
private static final String SEGMENT_NAME_PARTITION_1 = "table__1__32__12345";
+ private static final String DECODER_CLASS =
"org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder";
+ // A distinct decoder for partition 1 fixtures so the passthrough assertion
catches a cross-wired/hardcoded value.
+ private static final String DECODER_CLASS_ALT =
"org.apache.pinot.plugin.inputformat.avro.KafkaAvroMessageDecoder";
private static final int TIMEOUT_MSEC = 10000;
private static final int EXTENDED_TIMEOUT_FACTOR = 100;
@@ -89,10 +92,10 @@ public class ConsumingSegmentInfoReaderStatelessTest {
FakeConsumingInfoServer s0 = new
FakeConsumingInfoServer(Lists.newArrayList(
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_0, "CONSUMING", 0,
partitionToOffset0,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset0,
Map.of(),
- Map.of(), Map.of())),
+ Map.of(), Map.of()), DECODER_CLASS),
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_1, "CONSUMING", 0,
partitionToOffset1,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset1,
Map.of(),
- Map.of(), Map.of()))));
+ Map.of(), Map.of()), DECODER_CLASS_ALT)));
s0.start(uriPath, createHandler(200, s0._consumerInfos, 0));
_serverMap.put("server0", s0);
@@ -100,10 +103,10 @@ public class ConsumingSegmentInfoReaderStatelessTest {
FakeConsumingInfoServer s1 = new
FakeConsumingInfoServer(Lists.newArrayList(
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_0, "CONSUMING", 0,
partitionToOffset0,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset0,
Map.of(),
- Map.of(), Map.of())),
+ Map.of(), Map.of()), DECODER_CLASS),
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_1, "CONSUMING", 0,
partitionToOffset1,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset1,
Map.of(),
- Map.of(), Map.of()))));
+ Map.of(), Map.of()), DECODER_CLASS_ALT)));
s1.start(uriPath, createHandler(200, s1._consumerInfos, 0));
_serverMap.put("server1", s1);
@@ -111,10 +114,10 @@ public class ConsumingSegmentInfoReaderStatelessTest {
FakeConsumingInfoServer s2 = new
FakeConsumingInfoServer(Lists.newArrayList(
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_0, "NOT_CONSUMING", 0,
partitionToOffset0,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset0,
Map.of(),
- Map.of(), Map.of())),
+ Map.of(), Map.of()), DECODER_CLASS),
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_1, "CONSUMING", 0,
partitionToOffset1,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset1,
Map.of(),
- Map.of(), Map.of()))));
+ Map.of(), Map.of()), DECODER_CLASS_ALT)));
s2.start(uriPath, createHandler(200, s2._consumerInfos, 0));
_serverMap.put("server2", s2);
@@ -122,7 +125,7 @@ public class ConsumingSegmentInfoReaderStatelessTest {
FakeConsumingInfoServer s3 = new
FakeConsumingInfoServer(Lists.newArrayList(
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_1, "CONSUMING", 0,
partitionToOffset1,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset1,
Map.of(),
- Map.of(), Map.of()))));
+ Map.of(), Map.of()), DECODER_CLASS_ALT)));
s3.start(uriPath, createHandler(200, s3._consumerInfos, 0));
_serverMap.put("server3", s3);
@@ -130,10 +133,10 @@ public class ConsumingSegmentInfoReaderStatelessTest {
FakeConsumingInfoServer s4 = new
FakeConsumingInfoServer(Lists.newArrayList(
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_0, "CONSUMING", 0,
partitionToOffset0,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset0,
Map.of(),
- Map.of(), Map.of())),
+ Map.of(), Map.of()), DECODER_CLASS),
new SegmentConsumerInfo(SEGMENT_NAME_PARTITION_1, "CONSUMING", 0,
partitionToOffset1,
new SegmentConsumerInfo.PartitionOffsetInfo(partitionToOffset1,
Map.of(),
- Map.of(), Map.of()))));
+ Map.of(), Map.of()), DECODER_CLASS_ALT)));
s4.start(uriPath, createHandler(200, s4._consumerInfos, TIMEOUT_MSEC *
EXTENDED_TIMEOUT_FACTOR));
_serverMap.put("server4", s4);
}
@@ -352,5 +355,52 @@ public class ConsumingSegmentInfoReaderStatelessTest {
assertTrue(serverNames.contains(info._serverName));
assertEquals(info._consumerState, consumerState);
assertEquals(info._partitionOffsetInfo._currentOffsetsMap.get(partition),
offset);
+ // decoderClassName from the server-side DTO must survive the
controller-aggregate reader passthrough. Partition 0
+ // and 1 fixtures use different decoders, so this also catches a
cross-wired or hardcoded value.
+ assertEquals(info._decoderClassName, partition.equals("0") ? DECODER_CLASS
: DECODER_CLASS_ALT);
+ }
+
+ /// An older server may return a payload without the decoderClassName field;
it must deserialize to null rather
+ /// than failing, preserving rolling-upgrade compatibility.
+ @Test
+ public void testDecoderClassNameMissingDeserializesToNull()
+ throws IOException {
+ SegmentConsumerInfo info = JsonUtils.stringToObject(
+
"{\"segmentName\":\"seg\",\"consumerState\":\"CONSUMING\",\"lastConsumedTimestamp\":0,"
+ + "\"partitionToOffsetMap\":{}}", SegmentConsumerInfo.class);
+ assertNull(info.getDecoderClassName());
+ }
+
+ /// The pre-decoderClassName `SegmentConsumerInfo` constructor must remain
source/binary compatible for DTO builders
+ /// outside this repo, and delegate with a null decoder class name.
+ @Test
+ public void testLegacyConstructorDelegatesWithNullDecoderClassName() {
+ SegmentConsumerInfo.PartitionOffsetInfo offsetInfo =
+ new SegmentConsumerInfo.PartitionOffsetInfo(Map.of(), Map.of(),
Map.of(), Map.of());
+ assertNull(new SegmentConsumerInfo("seg", "CONSUMING", 0, Map.of(),
offsetInfo).getDecoderClassName());
+ }
+
+ /// Jackson must resolve exactly one creator for each DTO
(`SegmentConsumerInfo` also has a legacy constructor) and
+ /// round-trip the decoder class name.
+ @Test
+ public void testDecoderClassNameJsonRoundTrip()
+ throws IOException {
+ SegmentConsumerInfo serverInfo =
JsonUtils.stringToObject(JsonUtils.objectToString(
+ new SegmentConsumerInfo("seg", "CONSUMING", 0, Map.of(),
+ new SegmentConsumerInfo.PartitionOffsetInfo(Map.of(), Map.of(),
Map.of(), Map.of()), DECODER_CLASS)),
+ SegmentConsumerInfo.class);
+ assertEquals(serverInfo.getDecoderClassName(), DECODER_CLASS);
+
+ ConsumingSegmentInfoReader.ConsumingSegmentInfo controllerInfo =
JsonUtils.stringToObject(JsonUtils.objectToString(
+ new ConsumingSegmentInfoReader.ConsumingSegmentInfo("server0",
"CONSUMING", 0, Map.of(),
+ new ConsumingSegmentInfoReader.PartitionOffsetInfo(Map.of(),
Map.of(), Map.of(), Map.of()),
+ DECODER_CLASS)),
ConsumingSegmentInfoReader.ConsumingSegmentInfo.class);
+ assertEquals(controllerInfo._decoderClassName, DECODER_CLASS);
+
+ // A payload from an older server (no decoderClassName) must still
deserialize through the same creator.
+ ConsumingSegmentInfoReader.ConsumingSegmentInfo legacyPayload =
JsonUtils.stringToObject(
+
"{\"serverName\":\"server0\",\"consumerState\":\"CONSUMING\",\"lastConsumedTimestamp\":0}",
+ ConsumingSegmentInfoReader.ConsumingSegmentInfo.class);
+ assertNull(legacyPayload._decoderClassName);
}
}
diff --git
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
index ccf8d7a805f..7298ecd5171 100644
---
a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
+++
b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/RealtimeConsumerMonitorTest.java
@@ -156,7 +156,7 @@ public class RealtimeConsumerMonitorTest {
new ConsumingSegmentInfoReader.PartitionOffsetInfo(currentOffsetMap,
latestUpstreamOffsetMap, recordsLagMap,
availabilityLagMsMap);
return new ConsumingSegmentInfoReader.ConsumingSegmentInfo(serverName,
"CONSUMING", -1, currentOffsetMap,
- partitionOffsetInfo);
+ partitionOffsetInfo, null);
}
Map<String, String> getStreamConfigMap() {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
index 8641227c9cd..c3b76c4fd6d 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java
@@ -253,6 +253,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
private final TableConfig _tableConfig;
private final RealtimeTableDataManager _realtimeTableDataManager;
private final StreamDataDecoder _streamDataDecoder;
+ private final String _decoderClassName;
private final int _segmentMaxRowCount;
private final String _resourceDataDir;
private final Schema _schema;
@@ -1629,6 +1630,12 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
}
}
+ /// @return the FQCN of the [StreamMessageDecoder] actually instantiated on
this consumer (the source of truth for
+ /// what is decoding this segment)
+ public String getDecoderClassName() {
+ return _decoderClassName;
+ }
+
/// Cleans up the metrics that reflects the state of the realtime segment.
/// This step is essential as the instance may not be the target location
for some of the partitions.
/// E.g. if the number of partitions increases, or a host swap is needed,
the target location for some partitions
@@ -2166,12 +2173,14 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
Set<String> fieldsToRead =
IngestionUtils.getFieldsForRecordExtractor(_tableConfig, _schema);
RetryPolicy retryPolicy = RetryPolicies.exponentialBackoffRetryPolicy(5,
1000L, 1.2f);
AtomicReference<StreamDataDecoder> localStreamDataDecoder = new
AtomicReference<>();
+ AtomicReference<String> localDecoderClassName = new AtomicReference<>();
try {
retryPolicy.attempt(() -> {
try {
StreamMessageDecoder streamMessageDecoder =
createMessageDecoder(fieldsToRead);
boolean isKeyBytesType =
StreamDataDecoderImpl.isKeyBytesType(_schema);
localStreamDataDecoder.set(new
StreamDataDecoderImpl(streamMessageDecoder, isKeyBytesType));
+ localDecoderClassName.set(streamMessageDecoder.getClass().getName());
return true;
} catch (Exception e) {
_segmentLogger.warn("Failed to initialize the StreamMessageDecoder:
", e);
@@ -2184,6 +2193,7 @@ public class RealtimeSegmentDataManager extends
SegmentDataManager {
throw e;
}
_streamDataDecoder = localStreamDataDecoder.get();
+ _decoderClassName = localDecoderClassName.get();
try {
_transformPipeline = new TransformPipeline(tableConfig, schema);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
index 49daefd1ff2..6237395b23f 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java
@@ -667,6 +667,15 @@ public class RealtimeSegmentDataManagerTest {
_partitionGroupIdToConsumerCoordinatorMap, serverMetrics,
timeSupplier);
}
+ @Test
+ public void testGetDecoderClassName()
+ throws Exception {
+ try (FakeRealtimeSegmentDataManager segmentDataManager =
createFakeSegmentManager()) {
+ // Reports the actual instantiated decoder class, not the one configured
on the table
+ assertEquals(segmentDataManager.getDecoderClassName(),
FakeStreamMessageDecoder.class.getName());
+ }
+ }
+
@BeforeClass
public void setUp() {
ServerMetrics.register(mock(ServerMetrics.class));
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DebugResource.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DebugResource.java
index 4b4e0832ca3..7aa588a25f0 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DebugResource.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/DebugResource.java
@@ -240,7 +240,8 @@ public class DebugResource {
realtimeSegmentDataManager.getLastConsumedTimestamp(),
currentOffsets,
new SegmentConsumerInfo.PartitionOffsetInfo(currentOffsets,
- upstreamLatest, recordsLagMap, availabilityLagMsMap));
+ upstreamLatest, recordsLagMap, availabilityLagMsMap),
+ realtimeSegmentDataManager.getDecoderClassName());
}
return segmentConsumerInfo;
}
diff --git
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
index 82bfa64f570..cb2bf048e89 100644
---
a/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
+++
b/pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java
@@ -1190,7 +1190,8 @@ public class TablesResource {
new
SegmentConsumerInfo.PartitionOffsetInfo(partitiionToOffsetMap,
partitionIdToStateMap.entrySet()
.stream()
.collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().getUpstreamLatestOffset().toString())),
- recordsLagMap, availabilityLagMsMap)));
+ recordsLagMap, availabilityLagMsMap),
+ realtimeSegmentDataManager.getDecoderClassName()));
}
}
} catch (Exception e) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
