Add DynamicConfigProvider for Schema Registry

core/src/main/java/org/apache/druid/utils/DynamicConfigProviderUtils.java

@@ -0,0 +1,73 @@
+/*
+ * 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.druid.utils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.metadata.DynamicConfigProvider;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class DynamicConfigProviderUtils
+{
+  public static Map<String, String> extraConfigAndSetStringMap(Map<String, Object> config, String dynamicConfigProviderKey, ObjectMapper mapper)
+  {
+    HashMap<String, String> newConfig = new HashMap<>();
+    if (config != null) {
+      for (Map.Entry<String, Object> entry : config.entrySet()) {
+        if (!dynamicConfigProviderKey.equals(entry.getKey())) {
+          newConfig.put(entry.getKey(), entry.getValue().toString());
+        }
+      }
+      Map<String, String> dynamicConfig = extraConfigFromProvider(config.get(dynamicConfigProviderKey), mapper);
+      for (Map.Entry<String, String> entry : dynamicConfig.entrySet()) {
+        newConfig.put(entry.getKey(), entry.getValue());
+      }
+    }
+    return newConfig;
+  }
+
+  public static Map<String, Object> extraConfigAndSetObjectMap(Map<String, Object> config, String dynamicConfigProviderKey, ObjectMapper mapper)
+  {
+    HashMap<String, Object> newConfig = new HashMap<>();
+    if (config != null) {
+      for (Map.Entry<String, Object> entry : config.entrySet()) {
+        if (!dynamicConfigProviderKey.equals(entry.getKey())) {
+          newConfig.put(entry.getKey(), entry.getValue());
+        }
+      }
+      Map<String, String> dynamicConfig = extraConfigFromProvider(config.get(dynamicConfigProviderKey), mapper);
+      for (Map.Entry<String, String> entry : dynamicConfig.entrySet()) {
+        newConfig.put(entry.getKey(), entry.getValue());
+      }
+    }
+    return newConfig;
+  }
+
+  private static Map<String, String> extraConfigFromProvider(Object dynamicConfigProviderJson, ObjectMapper mapper)
+  {
+    if (dynamicConfigProviderJson != null) {
+      DynamicConfigProvider dynamicConfigProvider = mapper.convertValue(dynamicConfigProviderJson, DynamicConfigProvider.class);
+      return dynamicConfigProvider.getConfig();
+    }
+    return Collections.emptyMap();
+  }
+}

core/src/test/java/org/apache/druid/utils/DynamicConfigProviderUtilsTest.java

@@ -0,0 +1,84 @@
+/*
+ * 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.druid.utils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.metadata.DynamicConfigProvider;
+import org.apache.druid.metadata.MapStringDynamicConfigProvider;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.runners.Enclosed;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+
+import java.util.Map;
+
+@RunWith(Enclosed.class)
+public class DynamicConfigProviderUtilsTest
+{
+  public static class ThrowIfURLHasNotAllowedPropertiesTest
+  {
+    @Rule
+    public ExpectedException expectedException = ExpectedException.none();
+
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+    private final String DYNAMIC_CONFIG_PROVIDER = "druid.dynamic.config.provider";
+
+    @Test
+    public void testExtraConfigAndSetStringMap()
+    {
+      DynamicConfigProvider dynamicConfigProvider = new MapStringDynamicConfigProvider(
+          ImmutableMap.of("prop2", "value2")
+      );
+
+      Map<String, Object> properties = ImmutableMap.of(
+          "prop1", "value1",
+          "prop2", "value3",
+          DYNAMIC_CONFIG_PROVIDER, OBJECT_MAPPER.convertValue(dynamicConfigProvider, Map.class)
+      );
+      Map<String, String> res = DynamicConfigProviderUtils.extraConfigAndSetStringMap(properties, DYNAMIC_CONFIG_PROVIDER, OBJECT_MAPPER);
+
+      Assert.assertEquals(2, res.size());
+      Assert.assertEquals("value1", res.get("prop1"));
+      Assert.assertEquals("value2", res.get("prop2"));
+    }
+
+    @Test
+    public void testExtraConfigAndSetObjectMap()
+    {
+      DynamicConfigProvider dynamicConfigProvider = new MapStringDynamicConfigProvider(
+          ImmutableMap.of("prop2", "value2")
+      );
+
+      Map<String, Object> properties = ImmutableMap.of(
+          "prop1", "value1",
+          "prop2", "value3",
+          DYNAMIC_CONFIG_PROVIDER, OBJECT_MAPPER.convertValue(dynamicConfigProvider, Map.class)
+      );
+      Map<String, Object> res = DynamicConfigProviderUtils.extraConfigAndSetObjectMap(properties, DYNAMIC_CONFIG_PROVIDER, OBJECT_MAPPER);
+
+      Assert.assertEquals(2, res.size());
+      Assert.assertEquals("value1", res.get("prop1").toString());
+      Assert.assertEquals("value2", res.get("prop2").toString());
+    }
+  }
+}

docs/ingestion/data-formats.md

@@ -380,8 +380,8 @@ For details, see the Schema Registry [documentation](http://docs.confluent.io/cu
 | url | String | Specifies the url endpoint of the Schema Registry. | yes |
 | capacity | Integer | Specifies the max size of the cache (default = Integer.MAX_VALUE). | no |
 | urls | Array<String> | Specifies the url endpoints of the multiple Schema Registry instances. | yes(if `url` is not provided) |
-| config | Json | To send additional configurations, configured for Schema Registry | no |
-| headers | Json | To send headers to the Schema Registry | no |
+| config | Json | To send additional configurations, configured for Schema Registry.  This can be supplied via a [DynamicConfigProvider](../operations/dynamic-config-provider.md) | no |
+| headers | Json | To send headers to the Schema Registry.  This can be supplied via a [DynamicConfigProvider](../operations/dynamic-config-provider.md) | no |
 
 For a single schema registry instance, use Field `url` or `urls` for multi instances.
 
@@ -408,12 +408,20 @@ Multiple Instances:
         "schema.registry.ssl.truststore.password": "<password>",
         "schema.registry.ssl.keystore.location": "/some/secrets/kafka.client.keystore.jks",
         "schema.registry.ssl.keystore.password": "<password>",
-        "schema.registry.ssl.key.password": "<password>"
+        "schema.registry.ssl.key.password": "<password>",
+        "schema.registry.ssl.key.password",
        ... 
    },
    "headers": {
        "traceID" : "b29c5de2-0db4-490b-b421",
        "timeStamp" : "1577191871865",
+       "druid.dynamic.config.provider":{
+            "type":"mapString", 
+            "config":{
+                 "registry.header.prop.1":"value.1", 
+                 "registry.header.prop.2":"value.2"
+                 }
+            }
        ...
     }
 }
@@ -1223,8 +1231,8 @@ For details, see the Schema Registry [documentation](http://docs.confluent.io/cu
 | url | String | Specifies the url endpoint of the Schema Registry. | yes |
 | capacity | Integer | Specifies the max size of the cache (default = Integer.MAX_VALUE). | no |
 | urls | Array<String> | Specifies the url endpoints of the multiple Schema Registry instances. | yes(if `url` is not provided) |
-| config | Json | To send additional configurations, configured for Schema Registry | no |
-| headers | Json | To send headers to the Schema Registry | no |
+| config | Json | To send additional configurations, configured for Schema Registry. This can be supplied via a [DynamicConfigProvider](../operations/dynamic-config-provider.md).  | no |
+| headers | Json | To send headers to the Schema Registry.  This can be supplied via a [DynamicConfigProvider](../operations/dynamic-config-provider.md) | no |
 
 For a single schema registry instance, use Field `url` or `urls` for multi instances.
 
@@ -1259,6 +1267,13 @@ Multiple Instances:
   "headers": {
       "traceID" : "b29c5de2-0db4-490b-b421",
       "timeStamp" : "1577191871865",
+      "druid.dynamic.config.provider":{
+           "type":"mapString", 
+           "config":{
+                "registry.header.prop.1":"value.1", 
+                "registry.header.prop.2":"value.2"
+                }
+           }
       ...
   }
 }

extensions-core/avro-extensions/src/main/java/org/apache/druid/data/input/avro/SchemaRegistryBasedAvroBytesDecoder.java

@@ -19,8 +19,10 @@
 
 package org.apache.druid.data.input.avro;
 
+import com.fasterxml.jackson.annotation.JacksonInject;
 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 io.confluent.kafka.schemaregistry.ParsedSchema;
 import io.confluent.kafka.schemaregistry.avro.AvroSchema;
@@ -32,8 +34,10 @@
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.DatumReader;
 import org.apache.avro.io.DecoderFactory;
+import org.apache.druid.guice.annotations.Json;
 import org.apache.druid.java.util.common.RE;
 import org.apache.druid.java.util.common.parsers.ParseException;
+import org.apache.druid.utils.DynamicConfigProviderUtils;
 
 import javax.annotation.Nullable;
 import java.io.IOException;
@@ -48,27 +52,31 @@ public class SchemaRegistryBasedAvroBytesDecoder implements AvroBytesDecoder
   private final String url;
   private final int capacity;
   private final List<String> urls;
-  private final Map<String, ?> config;
-  private final Map<String, String> headers;
+  private final Map<String, Object> config;
+  private final Map<String, Object> headers;
+  private final ObjectMapper jsonMapper;
+  public static final String DRUID_DYNAMIC_CONFIG_PROVIDER_KEY = "druid.dynamic.config.provider";
 
   @JsonCreator
   public SchemaRegistryBasedAvroBytesDecoder(
       @JsonProperty("url") @Deprecated String url,
       @JsonProperty("capacity") Integer capacity,
       @JsonProperty("urls") @Nullable List<String> urls,
-      @JsonProperty("config") @Nullable Map<String, ?> config,
-      @JsonProperty("headers") @Nullable Map<String, String> headers
+      @JsonProperty("config") @Nullable Map<String, Object> config,
+      @JsonProperty("headers") @Nullable Map<String, Object> headers,
+      @JacksonInject @Json ObjectMapper jsonMapper
   )
   {
     this.url = url;
     this.capacity = capacity == null ? Integer.MAX_VALUE : capacity;
     this.urls = urls;
     this.config = config;
     this.headers = headers;
+    this.jsonMapper = jsonMapper;
     if (url != null && !url.isEmpty()) {
-      this.registry = new CachedSchemaRegistryClient(this.url, this.capacity, this.config, this.headers);
+      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, this.config, this.headers);
+      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));
     }
   }
 
@@ -91,13 +99,13 @@ public List<String> getUrls()
   }
 
   @JsonProperty
-  public Map<String, ?> getConfig()
+  public Map<String, Object> getConfig()
   {
     return config;
   }
 
   @JsonProperty
-  public Map<String, String> getHeaders()
+  public Map<String, Object> getHeaders()
   {
     return headers;
   }
@@ -112,6 +120,7 @@ public Map<String, String> getHeaders()
     this.config = null;
     this.headers = null;
     this.registry = registry;
+    this.jsonMapper = new ObjectMapper();
   }
 
   @Override

extensions-core/avro-extensions/src/test/java/org/apache/druid/data/input/AvroStreamInputFormatTest.java

@@ -19,6 +19,7 @@
 
 package org.apache.druid.data.input;
 
+import com.fasterxml.jackson.databind.InjectableValues;
 import com.fasterxml.jackson.databind.Module;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
@@ -109,6 +110,9 @@ public void before()
     for (Module jacksonModule : new AvroExtensionsModule().getJacksonModules()) {
       jsonMapper.registerModule(jacksonModule);
     }
+    jsonMapper.setInjectableValues(
+        new InjectableValues.Std().addValue(ObjectMapper.class, new DefaultObjectMapper())
+    );
   }
 
   @Test
@@ -134,7 +138,7 @@ public void testSerdeForSchemaRegistry() throws IOException
   {
     AvroStreamInputFormat inputFormat = new AvroStreamInputFormat(
         flattenSpec,
-        new SchemaRegistryBasedAvroBytesDecoder("http://test:8081", 100, null, null, null),
+        new SchemaRegistryBasedAvroBytesDecoder("http://test:8081", 100, null, null, null, null),
         false,
         false
     );