KAFKA-7509: Clean up incorrect warnings logged by Connect

clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java

@@ -24,6 +24,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.stream.Stream;
 
 /**
  * Configurations shared by Kafka client applications: producer, consumer, connect, etc.
@@ -182,6 +183,15 @@ public class CommonClientConfigs {
     public static final String DEFAULT_API_TIMEOUT_MS_DOC = "Specifies the timeout (in milliseconds) for client APIs. " +
             "This configuration is used as the default timeout for all client operations that do not specify a <code>timeout</code> parameter.";
 
+    public static final String CONNECT_KAFKA_CLUSTER_ID = "connect.kafka.cluster.id";
+    public static final String CONNECT_GROUP_ID = "connect.group.id";
+
+    public static void ignoreAutoPopulatedMetricsContextProperties(AbstractConfig config) {
+        Stream.of(CONNECT_KAFKA_CLUSTER_ID, CONNECT_GROUP_ID)
+                .map(property -> METRICS_CONTEXT_PREFIX + property)
+                .forEach(config::ignore);
+    }
+
     /**
      * Postprocess the configuration so that exponential backoff is disabled when reconnect backoff
      * is explicitly configured but the maximum reconnect backoff is not explicitly configured.

clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.SecurityConfig;
+import org.apache.kafka.common.config.internals.RecordingMap;
 import org.apache.kafka.common.metrics.Sensor;
 
 import java.util.Map;
@@ -218,6 +219,8 @@ public class AdminClientConfig extends AbstractConfig {
                                 .withClientSaslSupport();
     }
 
+    final boolean isSubConfig;
+
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
         return CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
@@ -229,6 +232,7 @@ public AdminClientConfig(Map<?, ?> props) {
 
     protected AdminClientConfig(Map<?, ?> props, boolean doLog) {
         super(CONFIG, props, doLog);
+        this.isSubConfig = props instanceof RecordingMap;
     }
 
     public static Set<String> configNames() {

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -592,7 +592,9 @@ private KafkaAdminClient(AdminClientConfig config,
             new TimeoutProcessorFactory() : timeoutProcessorFactory;
         this.maxRetries = config.getInt(AdminClientConfig.RETRIES_CONFIG);
         this.retryBackoffMs = config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG);
-        config.logUnused();
+        CommonClientConfigs.ignoreAutoPopulatedMetricsContextProperties(config);
+        if (!config.isSubConfig)
+            config.logUnused();
         AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
         log.debug("Kafka admin client initialized");
         thread.start();

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -24,14 +24,14 @@
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.SecurityConfig;
+import org.apache.kafka.common.config.internals.RecordingMap;
 import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.requests.JoinGroupRequest;
 import org.apache.kafka.common.serialization.Deserializer;
 
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -577,6 +577,8 @@ public class ConsumerConfig extends AbstractConfig {
                                 .withClientSaslSupport();
     }
 
+    final boolean isSubConfig;
+
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
         Map<String, Object> refinedConfigs = CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
@@ -601,7 +603,7 @@ private void maybeOverrideClientId(Map<String, Object> configs) {
     protected static Map<String, Object> appendDeserializerToConfig(Map<String, Object> configs,
                                                                     Deserializer<?> keyDeserializer,
                                                                     Deserializer<?> valueDeserializer) {
-        Map<String, Object> newConfigs = new HashMap<>(configs);
+        Map<String, Object> newConfigs = RecordingMap.copyAndPreserve(configs);
         if (keyDeserializer != null)
             newConfigs.put(KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer.getClass());
         if (valueDeserializer != null)
@@ -624,14 +626,17 @@ boolean maybeOverrideEnableAutoCommit() {
 
     public ConsumerConfig(Properties props) {
         super(CONFIG, props);
+        this.isSubConfig = false;
     }
 
     public ConsumerConfig(Map<String, Object> props) {
         super(CONFIG, props);
+        this.isSubConfig = props instanceof RecordingMap;
     }
 
     protected ConsumerConfig(Map<?, ?> props, boolean doLog) {
         super(CONFIG, props, doLog);
+        this.isSubConfig = props instanceof RecordingMap;
     }
 
     public static Set<String> configNames() {

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -812,7 +812,10 @@ public KafkaConsumer(Map<String, Object> configs,
 
             this.kafkaConsumerMetrics = new KafkaConsumerMetrics(metrics, metricGrpPrefix);
 
-            config.logUnused();
+            CommonClientConfigs.ignoreAutoPopulatedMetricsContextProperties(config);
+
+            if (!config.isSubConfig)
+                config.logUnused();
             AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
             log.debug("Kafka consumer initialized");
         } catch (Throwable t) {

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -429,7 +429,9 @@ public KafkaProducer(Properties properties, Serializer<K> keySerializer, Seriali
             String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
             this.ioThread = new KafkaThread(ioThreadName, this.sender, true);
             this.ioThread.start();
-            config.logUnused();
+            CommonClientConfigs.ignoreAutoPopulatedMetricsContextProperties(config);
+            if (!config.isSubConfig)
+                config.logUnused();
             AppInfoParser.registerAppInfo(JMX_PREFIX, clientId, metrics, time.milliseconds());
             log.debug("Kafka producer started");
         } catch (Throwable t) {

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -25,13 +25,13 @@
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.SecurityConfig;
+import org.apache.kafka.common.config.internals.RecordingMap;
 import org.apache.kafka.common.metrics.Sensor;
 import org.apache.kafka.common.serialization.Serializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
@@ -453,6 +453,8 @@ public class ProducerConfig extends AbstractConfig {
                                         TRANSACTIONAL_ID_DOC);
     }
 
+    final boolean isSubConfig;
+
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
         Map<String, Object> refinedConfigs = CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
@@ -537,7 +539,7 @@ private static String parseAcks(String acksString) {
     static Map<String, Object> appendSerializerToConfig(Map<String, Object> configs,
             Serializer<?> keySerializer,
             Serializer<?> valueSerializer) {
-        Map<String, Object> newConfigs = new HashMap<>(configs);
+        Map<String, Object> newConfigs = RecordingMap.copyAndPreserve(configs);
         if (keySerializer != null)
             newConfigs.put(KEY_SERIALIZER_CLASS_CONFIG, keySerializer.getClass());
         if (valueSerializer != null)
@@ -547,14 +549,17 @@ static Map<String, Object> appendSerializerToConfig(Map<String, Object> configs,
 
     public ProducerConfig(Properties props) {
         super(CONFIG, props);
+        this.isSubConfig = false;
     }
 
     public ProducerConfig(Map<String, Object> props) {
         super(CONFIG, props);
+        this.isSubConfig = props instanceof RecordingMap;
     }
 
     ProducerConfig(Map<?, ?> props, boolean doLog) {
         super(CONFIG, props, doLog);
+        this.isSubConfig = props instanceof RecordingMap;
     }
 
     public static Set<String> configNames() {

clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.Configurable;
 import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.config.internals.RecordingMap;
 import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.utils.Utils;
 import org.slf4j.Logger;
@@ -227,13 +228,13 @@ public Set<String> unused() {
     }
 
     public Map<String, Object> originals() {
-        Map<String, Object> copy = new RecordingMap<>();
+        Map<String, Object> copy = new RecordingMap<>(this);
         copy.putAll(originals);
         return copy;
     }
 
     public Map<String, Object> originals(Map<String, Object> configOverrides) {
-        Map<String, Object> copy = new RecordingMap<>();
+        Map<String, Object> copy = new RecordingMap<>(this);
         copy.putAll(originals);
         copy.putAll(configOverrides);
         return copy;
@@ -245,7 +246,7 @@ public Map<String, Object> originals(Map<String, Object> configOverrides) {
      * @throws ClassCastException if any of the values are not strings
      */
     public Map<String, String> originalsStrings() {
-        Map<String, String> copy = new RecordingMap<>();
+        Map<String, String> copy = new RecordingMap<>(this);
         for (Map.Entry<String, ?> entry : originals.entrySet()) {
             if (!(entry.getValue() instanceof String))
                 throw new ClassCastException("Non-string value found in original settings for key " + entry.getKey() +
@@ -273,7 +274,7 @@ public Map<String, Object> originalsWithPrefix(String prefix) {
      * @return a Map containing the settings with the prefix
      */
     public Map<String, Object> originalsWithPrefix(String prefix, boolean strip) {
-        Map<String, Object> result = new RecordingMap<>(prefix, false);
+        Map<String, Object> result = new RecordingMap<>(this, prefix, false);
         for (Map.Entry<String, ?> entry : originals.entrySet()) {
             if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length()) {
                 if (strip)
@@ -302,7 +303,7 @@ public Map<String, Object> originalsWithPrefix(String prefix, boolean strip) {
      * </p>
      */
     public Map<String, Object> valuesWithPrefixOverride(String prefix) {
-        Map<String, Object> result = new RecordingMap<>(values(), prefix, true);
+        Map<String, Object> result = new RecordingMap<>(this, values(), prefix, true);
         for (Map.Entry<String, ?> entry : originals.entrySet()) {
             if (entry.getKey().startsWith(prefix) && entry.getKey().length() > prefix.length()) {
                 String keyWithNoPrefix = entry.getKey().substring(prefix.length());
@@ -331,9 +332,9 @@ public Map<String, Object> valuesWithPrefixAllOrNothing(String prefix) {
         Map<String, Object> withPrefix = originalsWithPrefix(prefix, true);
 
         if (withPrefix.isEmpty()) {
-            return new RecordingMap<>(values(), "", true);
+            return new RecordingMap<>(this, values(), "", true);
         } else {
-            Map<String, Object> result = new RecordingMap<>(prefix, true);
+            Map<String, Object> result = new RecordingMap<>(this, prefix, true);
 
             for (Map.Entry<String, ?> entry : withPrefix.entrySet()) {
                 ConfigDef.ConfigKey configKey = definition.configKeys().get(entry.getKey());
@@ -346,11 +347,11 @@ public Map<String, Object> valuesWithPrefixAllOrNothing(String prefix) {
     }
 
     public Map<String, ?> values() {
-        return new RecordingMap<>(values);
+        return new RecordingMap<>(this, values);
     }
 
     public Map<String, ?> nonInternalValues() {
-        Map<String, Object> nonInternalConfigs = new RecordingMap<>();
+        Map<String, Object> nonInternalConfigs = new RecordingMap<>(this);
         values.forEach((key, value) -> {
             ConfigDef.ConfigKey configKey = definition.configKeys().get(key);
             if (configKey == null || !configKey.internalConfig) {
@@ -382,7 +383,7 @@ private void logAll() {
     public void logUnused() {
         Set<String> unusedkeys = unused();
         if (!unusedkeys.isEmpty()) {
-            log.warn("These configurations '{}' were supplied but are not used yet.", unusedkeys);
+            log.warn("These configurations were supplied but are not used yet: {}", unusedkeys);
         }
     }
 
@@ -605,52 +606,6 @@ public int hashCode() {
         return originals.hashCode();
     }
 
-    /**
-     * Marks keys retrieved via `get` as used. This is needed because `Configurable.configure` takes a `Map` instead
-     * of an `AbstractConfig` and we can't change that without breaking public API like `Partitioner`.
-     */
-    private class RecordingMap<V> extends HashMap<String, V> {
-
-        private final String prefix;
-        private final boolean withIgnoreFallback;
-
-        RecordingMap() {
-            this("", false);
-        }
-
-        RecordingMap(String prefix, boolean withIgnoreFallback) {
-            this.prefix = prefix;
-            this.withIgnoreFallback = withIgnoreFallback;
-        }
-
-        RecordingMap(Map<String, ? extends V> m) {
-            this(m, "", false);
-        }
-
-        RecordingMap(Map<String, ? extends V> m, String prefix, boolean withIgnoreFallback) {
-            super(m);
-            this.prefix = prefix;
-            this.withIgnoreFallback = withIgnoreFallback;
-        }
-
-        @Override
-        public V get(Object key) {
-            if (key instanceof String) {
-                String stringKey = (String) key;
-                String keyWithPrefix;
-                if (prefix.isEmpty()) {
-                    keyWithPrefix = stringKey;
-                } else {
-                    keyWithPrefix = prefix + stringKey;
-                }
-                ignore(keyWithPrefix);
-                if (withIgnoreFallback)
-                    ignore(stringKey);
-            }
-            return super.get(key);
-        }
-    }
-
     /**
      * ResolvingMap keeps a track of the original map instance and the resolved configs.
      * The originals are tracked in a separate nested map and may be a `RecordingMap`; thus