Skip to content

Replace schema registry TODO with explicit redirect in ProtoBufMessageDecoder #18818

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lazyfetch wants to merge 3 commits into
base: master
Choose a base branch
from
+254 −249
Open

Replace schema registry TODO with explicit redirect in ProtoBufMessageDecoder #18818

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9a20cc4
Replace TODO with explicit schema registry guidance in ProtoBufMessag…
lazyfetch Jun 21, 2026
64cbbca
Fix indentation, update comment to Markdown style and remove redundan…
lazyfetch Jun 22, 2026
d36ca7b
Fix indentation
lazyfetch Jun 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 11 additions & 5 deletions ...uf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,10 @@
import org.apache.pinot.spi.stream.StreamMessageDecoder;


//TODO: Add support for Schema Registry
// This decoder is intentionally file-based and does not support schema registry.
// For Confluent Schema Registry-backed descriptor resolution, use
// KafkaConfluentSchemaRegistryProtoBufMessageDecoder instead.
Comment thread
lazyfetch marked this conversation as resolved.
Outdated

public class ProtoBufMessageDecoder implements StreamMessageDecoder<byte[]> {
public static final String DESCRIPTOR_FILE_PATH = "descriptorFile";
public static final String PROTO_CLASS_NAME = "protoClassName";
Expand All @@ -43,13 +46,16 @@ public class ProtoBufMessageDecoder implements StreamMessageDecoder<byte[]> {

@Override
public void init(Map<String, String> props, Set<String> fieldsToRead, String topicName)
throws Exception {
throws Exception {
Comment thread
lazyfetch marked this conversation as resolved.
Outdated
Preconditions.checkState(props.containsKey(DESCRIPTOR_FILE_PATH),
"Protocol Buffer schema descriptor file must be provided");
"Property '%s' must be specified for ProtoBufMessageDecoder. "
+ "If you are using Confluent Schema Registry, use "
+ "KafkaConfluentSchemaRegistryProtoBufMessageDecoder instead.",
DESCRIPTOR_FILE_PATH);

_protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
InputStream descriptorFileInputStream = ProtoBufUtils.getDescriptorFileInputStream(
props.get(DESCRIPTOR_FILE_PATH));
props.get(DESCRIPTOR_FILE_PATH));
Comment thread
lazyfetch marked this conversation as resolved.
Outdated
Descriptors.Descriptor descriptor = buildProtoBufDescriptor(descriptorFileInputStream);
_recordExtractor = new ProtoBufRecordExtractor();
_recordExtractor.init(fieldsToRead, null);
Expand All @@ -58,7 +64,7 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
}

private Descriptors.Descriptor buildProtoBufDescriptor(InputStream fin)
throws IOException {
throws IOException {
try {
DynamicSchema dynamicSchema = DynamicSchema.parseFrom(fin);

Expand Down
21 changes: 21 additions & 0 deletions ...rc/test/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import static org.apache.pinot.plugin.inputformat.protobuf.ProtoBufTestDataGenerator.*;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;


public class ProtoBufMessageDecoderTest {
Expand Down Expand Up @@ -154,4 +155,24 @@ public void testCompositeMessage()
assertEquals(((Map<String, Object>) destination.getValue("sample_record")).get("name"), "Alice");
assertEquals(((Map<String, Object>) destination.getValue("sample_record")).get("id"), 18);
}

@Test
public void testInitThrowsHelpfulMessageWhenDescriptorFileMissing() {
Comment thread
lazyfetch marked this conversation as resolved.
Outdated
ProtoBufMessageDecoder decoder = new ProtoBufMessageDecoder();
Map<String, String> props = new HashMap<>();
// No descriptorFile provided — should fail with a message pointing
// to KafkaConfluentSchemaRegistryProtoBufMessageDecoder
try {
decoder.init(props, ImmutableSet.of(), "test-topic");
throw new AssertionError("Expected IllegalStateException when descriptorFile is missing");
} catch (IllegalStateException e) {
assertNotNull(e.getMessage());
assertTrue(e.getMessage().contains("KafkaConfluentSchemaRegistryProtoBufMessageDecoder"),
"Error message should point users to the registry-backed decoder, but was: " + e.getMessage());
assertTrue(e.getMessage().contains(ProtoBufMessageDecoder.DESCRIPTOR_FILE_PATH),
"Error message should mention the missing property name, but was: " + e.getMessage());
} catch (Exception e) {
throw new AssertionError("Expected IllegalStateException but got: " + e.getClass().getName(), e);
}
}
}
Loading