This is an automated email from the ASF dual-hosted git repository.
cryptoe pushed a commit to branch 39.0.0
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/39.0.0 by this push:
new 4c0c1cfdf18 lazy_serde (#20789) (#20798)
4c0c1cfdf18 is described below
commit 4c0c1cfdf186b0bb7fdb2dcbf1c885bd7f951a48
Author: Karan Kumar <[email protected]>
AuthorDate: Fri Oct 30 05:59:49 2026 +0530
lazy_serde (#20789) (#20798)
---
.../druid/iceberg/input/GlueIcebergCatalog.java | 1 -
.../druid/iceberg/input/HiveIcebergCatalog.java | 4 +--
.../apache/druid/iceberg/input/LocalCatalog.java | 2 --
.../iceberg/input/HiveIcebergCatalogTest.java | 20 +++++++++++++++
.../avro/SchemaRegistryBasedAvroBytesDecoder.java | 21 +++++++++------
.../SchemaRegistryBasedAvroBytesDecoderTest.java | 15 +++++++++++
.../druid/inputsource/hdfs/HdfsInputSource.java | 10 +++-----
.../inputsource/hdfs/HdfsInputSourceTest.java | 14 ++++++++++
.../DescriptorBasedProtobufBytesDecoder.java | 30 +++++++++++++++-------
.../protobuf/FileBasedProtobufBytesDecoder.java | 3 +--
.../InlineDescriptorProtobufBytesDecoder.java | 3 +--
.../SchemaRegistryBasedProtobufBytesDecoder.java | 13 ++++++----
.../FileBasedProtobufBytesDecoderTest.java | 13 +++++-----
.../InlineDescriptorProtobufBytesDecoderTest.java | 15 +++++------
...chemaRegistryBasedProtobufBytesDecoderTest.java | 20 +++++++++++++++
15 files changed, 130 insertions(+), 54 deletions(-)
diff --git
a/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/GlueIcebergCatalog.java
b/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/GlueIcebergCatalog.java
index 5dfcebb062f..ccb9ff29681 100644
---
a/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/GlueIcebergCatalog.java
+++
b/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/GlueIcebergCatalog.java
@@ -78,7 +78,6 @@ public class GlueIcebergCatalog extends IcebergCatalog
mapper
);
this.caseSensitive = caseSensitive == null ? true : caseSensitive;
- this.catalog = retrieveCatalog();
}
@Override
diff --git
a/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/HiveIcebergCatalog.java
b/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/HiveIcebergCatalog.java
index f3e47ab32bd..ad86c784991 100644
---
a/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/HiveIcebergCatalog.java
+++
b/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/HiveIcebergCatalog.java
@@ -80,10 +80,10 @@ public class HiveIcebergCatalog extends IcebergCatalog
this.catalogUri = Preconditions.checkNotNull(catalogUri, "catalogUri
cannot be null");
this.catalogProperties =
DynamicConfigProviderUtils.extraConfigAndSetStringMap(catalogProperties,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, mapper);
this.caseSensitive = caseSensitive == null ? true : caseSensitive;
- this.configuration = configuration;
+ // Copy the injected configuration: it is shared by all catalogs, so
per-spec properties must not leak into it.
+ this.configuration = new Configuration(configuration);
this.catalogProperties
.forEach(this.configuration::set);
- this.hiveCatalog = retrieveCatalog();
}
@Override
diff --git
a/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/LocalCatalog.java
b/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/LocalCatalog.java
index 47198472474..80b8e3b334b 100644
---
a/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/LocalCatalog.java
+++
b/extensions-contrib/druid-iceberg-extensions/src/main/java/org/apache/druid/iceberg/input/LocalCatalog.java
@@ -60,8 +60,6 @@ public class LocalCatalog extends IcebergCatalog
this.warehousePath = warehousePath;
this.catalogProperties = catalogProperties;
this.caseSensitive = caseSensitive == null ? true : caseSensitive;
- this.catalog = retrieveCatalog();
-
}
@JsonProperty
diff --git
a/extensions-contrib/druid-iceberg-extensions/src/test/java/org/apache/druid/iceberg/input/HiveIcebergCatalogTest.java
b/extensions-contrib/druid-iceberg-extensions/src/test/java/org/apache/druid/iceberg/input/HiveIcebergCatalogTest.java
index 6c6489068d9..47cda038c7d 100644
---
a/extensions-contrib/druid-iceberg-extensions/src/test/java/org/apache/druid/iceberg/input/HiveIcebergCatalogTest.java
+++
b/extensions-contrib/druid-iceberg-extensions/src/test/java/org/apache/druid/iceberg/input/HiveIcebergCatalogTest.java
@@ -55,9 +55,29 @@ public class HiveIcebergCatalogTest
new Configuration()
);
Assertions.assertEquals("hive", hiveCatalog.retrieveCatalog().name());
+ Assertions.assertEquals(0,
hiveCatalogNullProps.getCatalogProperties().size());
+ hiveCatalogNullProps.retrieveCatalog();
Assertions.assertEquals(2,
hiveCatalogNullProps.getCatalogProperties().size());
}
+ @Test
+ public void testCatalogPropertiesDoNotLeakIntoSharedConfiguration()
+ {
+ final Configuration sharedConf = new Configuration();
+ HashMap<String, Object> catalogMap = new HashMap<>();
+ catalogMap.put("fs.defaultFS", "hdfs://leaked");
+ final HiveIcebergCatalog hiveCatalog = new HiveIcebergCatalog(
+ FileUtils.createTempDir().getPath(),
+ "hdfs://testuri",
+ catalogMap,
+ null,
+ mapper,
+ sharedConf
+ );
+ hiveCatalog.retrieveCatalog();
+ Assertions.assertNotEquals("hdfs://leaked",
sharedConf.get("fs.defaultFS"));
+ }
+
@Test
public void testAuthenticate()
{
diff --git
a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java
b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java
index c7c7438b1cd..1c3eaa105d6 100644
---
a/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java
+++
b/extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
import io.confluent.kafka.schemaregistry.ParsedSchema;
import io.confluent.kafka.schemaregistry.avro.AvroSchema;
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
@@ -47,7 +49,7 @@ import java.util.Objects;
public class SchemaRegistryBasedAvroBytesDecoder implements AvroBytesDecoder
{
- private final SchemaRegistryClient registry;
+ private final Supplier<SchemaRegistryClient> registry;
private final String url;
private final int capacity;
private final List<String> urls;
@@ -72,11 +74,14 @@ public class SchemaRegistryBasedAvroBytesDecoder implements
AvroBytesDecoder
this.config = config;
this.headers = headers;
this.jsonMapper = jsonMapper;
- if (url != null && !url.isEmpty()) {
- this.registry = new CachedSchemaRegistryClient(this.url, this.capacity,
DynamicConfigProviderUtils.extraConfigAndSetObjectMap(config,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper),
DynamicConfigProviderUtils.extraConfigAndSetStringMap(headers,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper));
- } else {
- this.registry = new CachedSchemaRegistryClient(this.urls, this.capacity,
DynamicConfigProviderUtils.extraConfigAndSetObjectMap(config,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper),
DynamicConfigProviderUtils.extraConfigAndSetStringMap(headers,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper));
- }
+ // Client is created lazily: its constructor reads keystores and other
resources named in the config.
+ this.registry = Suppliers.memoize(() -> {
+ if (url != null && !url.isEmpty()) {
+ return new CachedSchemaRegistryClient(this.url, this.capacity,
DynamicConfigProviderUtils.extraConfigAndSetObjectMap(config,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper),
DynamicConfigProviderUtils.extraConfigAndSetStringMap(headers,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper));
+ } else {
+ return new CachedSchemaRegistryClient(this.urls, this.capacity,
DynamicConfigProviderUtils.extraConfigAndSetObjectMap(config,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper),
DynamicConfigProviderUtils.extraConfigAndSetStringMap(headers,
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY, this.jsonMapper));
+ }
+ });
}
@JsonProperty
@@ -118,7 +123,7 @@ public class SchemaRegistryBasedAvroBytesDecoder implements
AvroBytesDecoder
this.urls = null;
this.config = null;
this.headers = null;
- this.registry = registry;
+ this.registry = () -> registry;
this.jsonMapper = new ObjectMapper();
}
@@ -136,7 +141,7 @@ public class SchemaRegistryBasedAvroBytesDecoder implements
AvroBytesDecoder
Schema schema;
try {
- ParsedSchema parsedSchema = registry.getSchemaById(id);
+ ParsedSchema parsedSchema = registry.get().getSchemaById(id);
schema = parsedSchema instanceof AvroSchema ? ((AvroSchema)
parsedSchema).rawSchema() : null;
}
catch (IOException ex1) {
diff --git
a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoderTest.java
b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoderTest.java
index ea21728ebd2..c9741983105 100644
---
a/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoderTest.java
+++
b/extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoderTest.java
@@ -111,6 +111,21 @@ public class SchemaRegistryBasedAvroBytesDecoderTest
Assertions.assertNotEquals(decoder.hashCode(), 0);
}
+ @Test
+ public void testClientNotCreatedOnDeserialization() throws Exception
+ {
+ // The truststore does not exist; creating the client would fail, so
deserialization must not create it.
+ String json = "{\"urls\":[\"https://localhost\"],\"type\":
\"schema_registry\", \"config\":"
+ +
"{\"schema.registry.ssl.truststore.location\":\"/nonexistent/truststore.jks\","
+ + "\"schema.registry.ssl.truststore.password\":\"x\"}}";
+ ObjectMapper mapper = new DefaultObjectMapper();
+ mapper.setInjectableValues(
+ new InjectableValues.Std().addValue(ObjectMapper.class, new
DefaultObjectMapper())
+ );
+
+ Assertions.assertDoesNotThrow(() ->
mapper.readerFor(AvroBytesDecoder.class).readValue(json));
+ }
+
@Test
public void testParse() throws Exception
{
diff --git
a/extensions-core/hdfs-storage/src/main/java/org/apache/druid/inputsource/hdfs/HdfsInputSource.java
b/extensions-core/hdfs-storage/src/main/java/org/apache/druid/inputsource/hdfs/HdfsInputSource.java
index 1ec082fb906..bb7897f93cc 100644
---
a/extensions-core/hdfs-storage/src/main/java/org/apache/druid/inputsource/hdfs/HdfsInputSource.java
+++
b/extensions-core/hdfs-storage/src/main/java/org/apache/druid/inputsource/hdfs/HdfsInputSource.java
@@ -128,13 +128,9 @@ public class HdfsInputSource
public static void verifyProtocol(Configuration conf, HdfsInputSourceConfig
config, String pathString)
{
- Path path = new Path(pathString);
- try {
- throwIfInvalidProtocol(config, path.getFileSystem(conf).getScheme());
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
+ // Resolve the scheme without creating a FileSystem, so a disallowed path
never touches the network.
+ final String scheme = new Path(pathString).toUri().getScheme();
+ throwIfInvalidProtocol(config, scheme != null ? scheme :
FileSystem.getDefaultUri(conf).getScheme());
}
private static void throwIfInvalidProtocol(HdfsInputSourceConfig config,
String scheme)
diff --git
a/extensions-core/hdfs-storage/src/test/java/org/apache/druid/inputsource/hdfs/HdfsInputSourceTest.java
b/extensions-core/hdfs-storage/src/test/java/org/apache/druid/inputsource/hdfs/HdfsInputSourceTest.java
index f88f30389b9..90c9b511df5 100644
---
a/extensions-core/hdfs-storage/src/test/java/org/apache/druid/inputsource/hdfs/HdfsInputSourceTest.java
+++
b/extensions-core/hdfs-storage/src/test/java/org/apache/druid/inputsource/hdfs/HdfsInputSourceTest.java
@@ -140,6 +140,20 @@ public class HdfsInputSourceTest extends
InitializedNullHandlingTest
assertTrue(exception.getMessage().contains("Only [druid] protocols are
allowed"));
}
+ @Test
+ public void testConstructorRejectsSchemeWithoutCreatingFileSystem()
+ {
+ // "nofs" has no FileSystem implementation; creating one would fail with
a different error.
+ Throwable exception = assertThrows(IllegalArgumentException.class, () ->
+ HdfsInputSource.builder()
+ .paths("nofs://somehost/foo/bar")
+ .configuration(CONFIGURATION)
+ .inputSourceConfig(DEFAULT_INPUT_SOURCE_CONFIG)
+ .build()
+ );
+ assertTrue(exception.getMessage().contains("Only [hdfs] protocols are
allowed"));
+ }
+
@Test
public void testConstructorWithDefaultHdfs()
{
diff --git
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/DescriptorBasedProtobufBytesDecoder.java
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/DescriptorBasedProtobufBytesDecoder.java
index a04ca012521..4911cea4631 100644
---
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/DescriptorBasedProtobufBytesDecoder.java
+++
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/DescriptorBasedProtobufBytesDecoder.java
@@ -25,7 +25,7 @@ import com.google.protobuf.ByteString;
import com.google.protobuf.DescriptorProtos;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
-import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.parsers.ParseException;
import javax.annotation.Nullable;
@@ -46,7 +46,7 @@ import java.util.TreeSet;
*/
public abstract class DescriptorBasedProtobufBytesDecoder implements
ProtobufBytesDecoder
{
- private Descriptors.Descriptor descriptor;
+ private volatile Descriptors.Descriptor descriptor;
/**
* An optional message Protobuf message type in the descriptor. Both short
name and fully qualified name are accepted.
@@ -71,15 +71,30 @@ public abstract class DescriptorBasedProtobufBytesDecoder
implements ProtobufByt
public Descriptors.Descriptor getDescriptor()
{
+ if (descriptor == null) {
+ initializeDescriptor();
+ }
return descriptor;
}
+ /**
+ * Loads the descriptor on first use. Subclasses may do I/O here, so it
should not be called from a
+ * {@link com.fasterxml.jackson.annotation.JsonCreator} constructor.
+ */
@VisibleForTesting
- void initializeDescriptor()
+ synchronized void initializeDescriptor()
{
if (this.descriptor == null) {
- final var descriptorSet = loadFileDescriptorSet();
- this.descriptor = buildMessageDescriptor(descriptorSet);
+ try {
+ final var descriptorSet = loadFileDescriptorSet();
+ this.descriptor = buildMessageDescriptor(descriptorSet);
+ }
+ catch (ParseException e) {
+ // Not a ParseException: a bad descriptor must fail the task, not be
counted as unparseable rows.
+ throw DruidException.forPersona(DruidException.Persona.OPERATOR)
+
.ofCategory(DruidException.Category.RUNTIME_FAILURE)
+ .build(e, "%s", e.getMessage());
+ }
}
}
@@ -89,10 +104,7 @@ public abstract class DescriptorBasedProtobufBytesDecoder
implements ProtobufByt
@Override
public DynamicMessage parse(ByteBuffer bytes)
{
- if (descriptor == null) {
- throw new IAE("Descriptor needs to be initialized before parsing");
- }
-
+ final Descriptors.Descriptor descriptor = getDescriptor();
try {
return DynamicMessage.parseFrom(descriptor, ByteString.copyFrom(bytes));
}
diff --git
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java
index 1f5b19ab0b6..64c5c32dd5c 100644
---
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java
+++
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoder.java
@@ -44,9 +44,8 @@ public class FileBasedProtobufBytesDecoder extends
DescriptorBasedProtobufBytesD
super(protoMessageType);
Preconditions.checkNotNull(descriptorFilePath);
+ // Descriptor is loaded lazily on first use, not during deserialization.
this.descriptorFilePath = descriptorFilePath;
-
- initializeDescriptor();
}
@JsonProperty("descriptor")
diff --git
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoder.java
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoder.java
index 3bc41b11241..0469d5a9688 100644
---
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoder.java
+++
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoder.java
@@ -44,8 +44,7 @@ public class InlineDescriptorProtobufBytesDecoder extends
DescriptorBasedProtobu
Preconditions.checkNotNull(descriptorString);
this.descriptorString = descriptorString;
-
- initializeDescriptor();
+ // Descriptor is loaded lazily on first use, not during deserialization.
}
@JsonProperty
diff --git
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoder.java
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoder.java
index 45bd0995270..b9a48cddb3e 100644
---
a/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoder.java
+++
b/extensions-core/protobuf-extensions/src/main/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoder.java
@@ -24,6 +24,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient;
@@ -52,7 +54,7 @@ public class SchemaRegistryBasedProtobufBytesDecoder
implements ProtobufBytesDec
private static final Logger LOGGER = new
Logger(SchemaRegistryBasedProtobufBytesDecoder.class);
- private final SchemaRegistryClient registry;
+ private final Supplier<SchemaRegistryClient> registry;
private final String url;
private final int capacity;
private final List<String> urls;
@@ -86,7 +88,8 @@ public class SchemaRegistryBasedProtobufBytesDecoder
implements ProtobufBytesDec
throw new IAE("SchemaRegistryBasedProtobufBytesDecoder requires non-null
URLs");
}
- this.registry = new CachedSchemaRegistryClient(
+ // Client is created lazily: its constructor reads keystores and other
resources named in the config.
+ this.registry = Suppliers.memoize(() -> new CachedSchemaRegistryClient(
baseUrls,
this.capacity,
Collections.singletonList(new ProtobufSchemaProvider()),
@@ -100,7 +103,7 @@ public class SchemaRegistryBasedProtobufBytesDecoder
implements ProtobufBytesDec
DRUID_DYNAMIC_CONFIG_PROVIDER_KEY,
this.jsonMapper
)
- );
+ ));
}
@JsonProperty
@@ -147,7 +150,7 @@ public class SchemaRegistryBasedProtobufBytesDecoder
implements ProtobufBytesDec
this.urls = null;
this.config = null;
this.headers = null;
- this.registry = registry;
+ this.registry = () -> registry;
this.jsonMapper = new ObjectMapper();
}
@@ -179,7 +182,7 @@ public class SchemaRegistryBasedProtobufBytesDecoder
implements ProtobufBytesDec
int length = bytes.limit() - 2 - 4;
Descriptors.Descriptor descriptor;
try {
- ProtobufSchema schema = (ProtobufSchema) registry.getSchemaById(id);
+ ProtobufSchema schema = (ProtobufSchema)
registry.get().getSchemaById(id);
descriptor = schema.toDescriptor();
}
catch (RestClientException e) {
diff --git
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoderTest.java
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoderTest.java
index 1e9b3fb9c88..dfac4032b34 100644
---
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoderTest.java
+++
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/FileBasedProtobufBytesDecoderTest.java
@@ -22,7 +22,7 @@ package org.apache.druid.data.input.protobuf;
import com.google.protobuf.Struct;
import com.google.protobuf.Value;
import com.google.protobuf.util.JsonFormat;
-import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.druid.error.DruidException;
import org.junit.jupiter.api.Test;
import java.io.File;
@@ -115,8 +115,8 @@ public class FileBasedProtobufBytesDecoderTest
public void testBadProto()
{
final var ex = assertThrows(
- ParseException.class,
- () -> new FileBasedProtobufBytesDecoder("proto_test_event.desc",
"BadName")
+ DruidException.class,
+ () -> new FileBasedProtobufBytesDecoder("proto_test_event.desc",
"BadName").getDescriptor()
);
assertEquals(
@@ -128,10 +128,9 @@ public class FileBasedProtobufBytesDecoderTest
@Test
public void testMalformedDescriptorUrl()
{
- final var ex = assertThrows(
- ParseException.class,
- () -> new FileBasedProtobufBytesDecoder("file:/nonexist.desc",
"BadName")
- );
+ // Construction must not touch the URL; the failure surfaces only on first
use.
+ final var decoder = assertDoesNotThrow(() -> new
FileBasedProtobufBytesDecoder("file:/nonexist.desc", "BadName"));
+ final var ex = assertThrows(DruidException.class, decoder::getDescriptor);
assertEquals(
"Failed to initialize descriptor at [file:/nonexist.desc]",
diff --git
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoderTest.java
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoderTest.java
index a38dbbb56b0..603f94c327c 100644
---
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoderTest.java
+++
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/InlineDescriptorProtobufBytesDecoderTest.java
@@ -20,9 +20,9 @@
package org.apache.druid.data.input.protobuf;
import com.google.common.io.Files;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.java.util.common.parsers.ParseException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -78,8 +78,8 @@ public class InlineDescriptorProtobufBytesDecoderTest
public void testBadProto()
{
final var ex = assertThrows(
- ParseException.class,
- () -> new InlineDescriptorProtobufBytesDecoder(descString, "BadName")
+ DruidException.class,
+ new InlineDescriptorProtobufBytesDecoder(descString,
"BadName")::getDescriptor
);
assertEquals(
@@ -93,7 +93,7 @@ public class InlineDescriptorProtobufBytesDecoderTest
{
final var ex = assertThrows(
IAE.class,
- () -> new InlineDescriptorProtobufBytesDecoder("invalidString",
"BadName")
+ new InlineDescriptorProtobufBytesDecoder("invalidString",
"BadName")::getDescriptor
);
assertEquals(
@@ -106,11 +106,8 @@ public class InlineDescriptorProtobufBytesDecoderTest
public void testMalformedDescriptorValidBase64InvalidDescriptor()
{
final var ex = assertThrows(
- ParseException.class,
- () -> new InlineDescriptorProtobufBytesDecoder(
- "aGVsbG8gd29ybGQ=",
- "BadName"
- )
+ DruidException.class,
+ new InlineDescriptorProtobufBytesDecoder("aGVsbG8gd29ybGQ=",
"BadName")::getDescriptor
);
assertEquals(
diff --git
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoderTest.java
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoderTest.java
index 1a5a27e1f04..309ba7727a3 100644
---
a/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoderTest.java
+++
b/extensions-core/protobuf-extensions/src/test/java/org/apache/druid/data/input/protobuf/SchemaRegistryBasedProtobufBytesDecoderTest.java
@@ -45,6 +45,7 @@ import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Map;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -116,6 +117,25 @@ public class SchemaRegistryBasedProtobufBytesDecoderTest
assertEquals(capacity,
schemaRegistryBasedProtobufBytesDecoder.getIdentityMapCapacity());
}
+ @Test
+ public void testClientNotCreatedOnConstruction()
+ {
+ // The truststore does not exist; creating the client would fail, so
construction must not create it.
+ final Map<String, Object> config = ImmutableMap.of(
+ "schema.registry.ssl.truststore.location",
"/nonexistent/truststore.jks",
+ "schema.registry.ssl.truststore.password", "x"
+ );
+ final var decoder = assertDoesNotThrow(() -> new
SchemaRegistryBasedProtobufBytesDecoder(
+ null,
+ null,
+ Collections.singletonList("https://localhost";),
+ config,
+ null,
+ new DefaultObjectMapper()
+ ));
+ assertEquals(config, decoder.getConfig());
+ }
+
private ProtoTestEventWrapper.ProtoTestEvent getTestEvent()
{
DateTime dateTime = new DateTime(2012, 7, 12, 9, 30,
ISOChronology.getInstanceUTC());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
