Replace schema registry TODO with explicit redirect in ProtoBufMessageDecoder

pinot-plugins/pinot-input-format/pinot-protobuf/src/main/java/org/apache/pinot/plugin/inputformat/protobuf/ProtoBufMessageDecoder.java

@@ -1,92 +1,96 @@
-/**
- * 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.plugin.inputformat.protobuf;
-
-import com.github.os72.protobuf.dynamic.DynamicSchema;
-import com.google.common.base.Preconditions;
-import com.google.protobuf.Descriptors;
-import com.google.protobuf.DynamicMessage;
-import com.google.protobuf.Message;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Map;
-import java.util.Set;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.pinot.spi.data.readers.GenericRow;
-import org.apache.pinot.spi.stream.StreamMessageDecoder;
-
-
-//TODO: Add support for Schema Registry
-public class ProtoBufMessageDecoder implements StreamMessageDecoder<byte[]> {
-  public static final String DESCRIPTOR_FILE_PATH = "descriptorFile";
-  public static final String PROTO_CLASS_NAME = "protoClassName";
-
-  private ProtoBufRecordExtractor _recordExtractor;
-  private String _protoClassName;
-  private Message.Builder _builder;
-
-  @Override
-  public void init(Map<String, String> props, Set<String> fieldsToRead, String topicName)
-      throws Exception {
-    Preconditions.checkState(props.containsKey(DESCRIPTOR_FILE_PATH),
-        "Protocol Buffer schema descriptor file must be provided");
-
-    _protoClassName = props.getOrDefault(PROTO_CLASS_NAME, "");
-    InputStream descriptorFileInputStream = ProtoBufUtils.getDescriptorFileInputStream(
-        props.get(DESCRIPTOR_FILE_PATH));
-    Descriptors.Descriptor descriptor = buildProtoBufDescriptor(descriptorFileInputStream);
-    _recordExtractor = new ProtoBufRecordExtractor();
-    _recordExtractor.init(fieldsToRead, null);
-    DynamicMessage dynamicMessage = DynamicMessage.getDefaultInstance(descriptor);
-    _builder = dynamicMessage.newBuilderForType();
-  }
-
-  private Descriptors.Descriptor buildProtoBufDescriptor(InputStream fin)
-      throws IOException {
-    try {
-      DynamicSchema dynamicSchema = DynamicSchema.parseFrom(fin);
-
-      if (!StringUtils.isEmpty(_protoClassName)) {
-        return dynamicSchema.getMessageDescriptor(_protoClassName);
-      } else {
-        return dynamicSchema.getMessageDescriptor(dynamicSchema.getMessageTypes().toArray(new String[]{})[0]);
-      }
-    } catch (Descriptors.DescriptorValidationException e) {
-      throw new IOException("Descriptor file validation failed", e);
-    }
-  }
-
-  @Override
-  public GenericRow decode(byte[] payload, GenericRow destination) {
-    return decode(payload, 0, payload.length, destination);
-  }
-
-  @Override
-  public GenericRow decode(byte[] payload, int offset, int length, GenericRow destination) {
-    Message message;
-    try {
-      message = _builder.mergeFrom(payload, offset, length).build();
-    } catch (Exception e) {
-      throw new RuntimeException("Caught exception while decoding protobuf message", e);
-    } finally {
-      _builder.clear();
-    }
-    return _recordExtractor.extract(message, destination);
-  }
-}
+/**
+ * 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.plugin.inputformat.protobuf;
+
+import com.github.os72.protobuf.dynamic.DynamicSchema;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
+import com.google.protobuf.Message;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Map;
+import java.util.Set;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.stream.StreamMessageDecoder;
+
+
+/// This decoder is intentionally file-based and does not support schema registry.
+/// For Confluent Schema Registry-backed descriptor resolution, use
+/// `KafkaConfluentSchemaRegistryProtoBufMessageDecoder` instead.
+
+public class ProtoBufMessageDecoder implements StreamMessageDecoder<byte[]> {
+  public static final String DESCRIPTOR_FILE_PATH = "descriptorFile";
+  public static final String PROTO_CLASS_NAME = "protoClassName";
+
+  private ProtoBufRecordExtractor _recordExtractor;
+  private String _protoClassName;
+  private Message.Builder _builder;
+
+  @Override
+  public void init(Map<String, String> props, Set<String> fieldsToRead, String topicName) throws Exception {
+    Preconditions.checkState(props.containsKey(DESCRIPTOR_FILE_PATH),
+            "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));
+    Descriptors.Descriptor descriptor = buildProtoBufDescriptor(descriptorFileInputStream);
+    _recordExtractor = new ProtoBufRecordExtractor();
+    _recordExtractor.init(fieldsToRead, null);
+    DynamicMessage dynamicMessage = DynamicMessage.getDefaultInstance(descriptor);
+    _builder = dynamicMessage.newBuilderForType();
+  }
+
+  private Descriptors.Descriptor buildProtoBufDescriptor(InputStream fin)
+          throws IOException {
+    try {
+      DynamicSchema dynamicSchema = DynamicSchema.parseFrom(fin);
+
+      if (!StringUtils.isEmpty(_protoClassName)) {
+        return dynamicSchema.getMessageDescriptor(_protoClassName);
+      } else {
+        return dynamicSchema.getMessageDescriptor(dynamicSchema.getMessageTypes().toArray(new String[]{})[0]);
+      }
+    } catch (Descriptors.DescriptorValidationException e) {
+      throw new IOException("Descriptor file validation failed", e);
+    }
+  }
+
+  @Override
+  public GenericRow decode(byte[] payload, GenericRow destination) {
+    return decode(payload, 0, payload.length, destination);
+  }
+
+  @Override
+  public GenericRow decode(byte[] payload, int offset, int length, GenericRow destination) {
+    Message message;
+    try {
+      message = _builder.mergeFrom(payload, offset, length).build();
+    } catch (Exception e) {
+      throw new RuntimeException("Caught exception while decoding protobuf message", e);
+    } finally {
+      _builder.clear();
+    }
+    return _recordExtractor.extract(message, destination);
+  }
+}