fix(protobuf-core): re-introduce the load-by-file-path/url protobuf descriptors

extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java

@@ -27,6 +27,8 @@
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URL;
 import java.util.Objects;
 
 public class FileBasedProtobufBytesDecoder extends DescriptorBasedProtobufBytesDecoder
@@ -56,20 +58,43 @@ public String getDescriptorFilePath()
   @Override
   protected DescriptorProtos.FileDescriptorSet loadFileDescriptorSet()
   {
-    try (InputStream fin = this.getClass().getClassLoader().getResourceAsStream(descriptorFilePath)) {
+    InputStream fin;
+    DescriptorProtos.FileDescriptorSet descriptorSet;
+    try {
+      fin = this.getClass().getClassLoader().getResourceAsStream(descriptorFilePath);
       if (fin == null) {
-        throw new ParseException(descriptorFilePath, "Descriptor not found in class path [%s]", descriptorFilePath);
+        URL url;
+        try {
+          url = new URL(descriptorFilePath);
+        }
+        catch (MalformedURLException e) {
+          throw new ParseException(
+              descriptorFilePath,
+              e,
+              "Descriptor not found in class path or malformed URL: [%s]", descriptorFilePath
+              );
+        }
+        try (InputStream urlIn = url.openConnection().getInputStream()) {
+          if (urlIn == null) {
+            throw new ParseException(
+                descriptorFilePath,
+                "Descriptor not found at URL: [%s]", descriptorFilePath
+            );
+          }
+          descriptorSet = DescriptorProtos.FileDescriptorSet.parseFrom(urlIn);
+        }
+      } else {
+        descriptorSet = DescriptorProtos.FileDescriptorSet.parseFrom(fin);
       }
 
-      final var descriptorSet = DescriptorProtos.FileDescriptorSet.parseFrom(fin);
       if (descriptorSet.getFileCount() == 0) {
         throw new ParseException(null, "No file descriptors found in the descriptor set");
       }
 
       return descriptorSet;
     }
     catch (IOException e) {
-      throw new ParseException(descriptorFilePath, e, "Failed to initialize descriptor");
+      throw new ParseException(descriptorFilePath, e, "Failed to initialize descriptor at [%s]", descriptorFilePath);
     }
   }
 

extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoderTest.java

@@ -25,6 +25,7 @@
 import org.apache.druid.java.util.common.parsers.ParseException;
 import org.junit.jupiter.api.Test;
 
+import java.io.File;
 import java.nio.ByteBuffer;
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -68,6 +69,16 @@ public void testMoreComplexProtoFile()
     assertEquals("prototest.ProtoNestedEvent", decoder.getDescriptor().getFullName());
   }
 
+  @Test
+  public void testDescriptorUrl()
+  {
+    File descFile = new File("src/test/resources/proto_test_event.desc");
+    String path = descFile.getAbsoluteFile().toString();
+
+    final var decoder = new FileBasedProtobufBytesDecoder("file://" + path, "ProtoTestEvent");
+    assertEquals("prototest.ProtoTestEvent", decoder.getDescriptor().getFullName());
+  }
+
   @Test
   public void testParsingWithMoreComplexProtoFile() throws Exception
   {
@@ -123,7 +134,7 @@ public void testMalformedDescriptorUrl()
     );
 
     assertEquals(
-        "Descriptor not found in class path [file:/nonexist.desc]",
+        "Failed to initialize descriptor at [file:/nonexist.desc]",
         ex.getMessage()
     );
   }