KAFKA-2372: Add Kafka-backed storage of Copycat configs.

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -26,9 +26,11 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.Properties;
@@ -419,6 +421,17 @@ public static Properties loadProps(String filename) throws IOException, FileNotF
         return props;
     }
 
+    /**
+     * Converts a Properties object to a Map<String, String>, calling {@link #toString} to ensure all keys and values
+     * are Strings.
+     */
+    public static Map<String, String> propsToStringMap(Properties props) {
+        Map<String, String> result = new HashMap<>();
+        for (Map.Entry<Object, Object> entry : props.entrySet())
+            result.put(entry.getKey().toString(), entry.getValue().toString());
+        return result;
+    }
+
     /**
      * Get the stack trace from an exception as a string
      */

config/copycat-distributed.properties

@@ -27,13 +27,14 @@ value.converter=org.apache.kafka.copycat.json.JsonConverter
 key.converter.schemas.enable=true
 value.converter.schemas.enable=true
 
-# The offset converter is configurable and must be specified, but most users will always want to use the built-in default.
-# Offset data is never visible outside of Copcyat.
-offset.key.converter=org.apache.kafka.copycat.json.JsonConverter
-offset.value.converter=org.apache.kafka.copycat.json.JsonConverter
-offset.key.converter.schemas.enable=false
-offset.value.converter.schemas.enable=false
+# The internal converter used for offsets and config data is configurable and must be specified, but most users will
+# always want to use the built-in default. Offset and config data is never visible outside of Copcyat in this format.
+internal.key.converter=org.apache.kafka.copycat.json.JsonConverter
+internal.value.converter=org.apache.kafka.copycat.json.JsonConverter
+internal.key.converter.schemas.enable=false
+internal.value.converter.schemas.enable=false
 
 offset.storage.topic=copycat-offsets
 # Flush much faster than normal, which is useful for testing/debugging
 offset.flush.interval.ms=10000
+config.storage.topic=copycat-configs

config/copycat-standalone.properties

@@ -25,12 +25,12 @@ value.converter=org.apache.kafka.copycat.json.JsonConverter
 key.converter.schemas.enable=true
 value.converter.schemas.enable=true
 
-# The offset converter is configurable and must be specified, but most users will always want to use the built-in default.
-# Offset data is never visible outside of Copcyat.
-offset.key.converter=org.apache.kafka.copycat.json.JsonConverter
-offset.value.converter=org.apache.kafka.copycat.json.JsonConverter
-offset.key.converter.schemas.enable=false
-offset.value.converter.schemas.enable=false
+# The internal converter used for offsets and config data is configurable and must be specified, but most users will
+# always want to use the built-in default. Offset and config data is never visible outside of Copcyat in this format.
+internal.key.converter=org.apache.kafka.copycat.json.JsonConverter
+internal.value.converter=org.apache.kafka.copycat.json.JsonConverter
+internal.key.converter.schemas.enable=false
+internal.value.converter.schemas.enable=false
 
 offset.storage.file.filename=/tmp/copycat.offsets
 # Flush much faster than normal, which is useful for testing/debugging

copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java

@@ -20,9 +20,8 @@
 import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.copycat.runtime.Copycat;
-import org.apache.kafka.copycat.runtime.Herder;
 import org.apache.kafka.copycat.runtime.Worker;
-import org.apache.kafka.copycat.runtime.standalone.StandaloneHerder;
+import org.apache.kafka.copycat.runtime.distributed.DistributedHerder;
 import org.apache.kafka.copycat.storage.KafkaOffsetBackingStore;
 import org.apache.kafka.copycat.util.Callback;
 import org.apache.kafka.copycat.util.FutureCallback;
@@ -59,7 +58,8 @@ public static void main(String[] args) throws Exception {
 
         WorkerConfig workerConfig = new WorkerConfig(workerProps);
         Worker worker = new Worker(workerConfig, new KafkaOffsetBackingStore());
-        Herder herder = new StandaloneHerder(worker);
+        DistributedHerder herder = new DistributedHerder(worker);
+        herder.configure(workerConfig.originals());
         final Copycat copycat = new Copycat(worker, herder);
         copycat.start();
 
@@ -73,7 +73,7 @@ public void onCompletion(Throwable error, String id) {
                             log.error("Failed to create job for {}", connectorPropsFile);
                     }
                 });
-                herder.addConnector(connectorProps, cb);
+                herder.addConnector(Utils.propsToStringMap(connectorProps), cb);
                 cb.get();
             }
         } catch (Throwable t) {

copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java

@@ -75,7 +75,7 @@ public void onCompletion(Throwable error, String id) {
                             log.error("Failed to create job for {}", connectorPropsFile);
                     }
                 });
-                herder.addConnector(connectorProps, cb);
+                herder.addConnector(Utils.propsToStringMap(connectorProps), cb);
                 cb.get();
             }
         } catch (Throwable t) {

copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java

@@ -57,13 +57,13 @@ public class WorkerConfig extends AbstractConfig {
     public static final String VALUE_CONVERTER_CLASS_DOC =
             "Converter class for value Copycat data that implements the <code>Converter</code> interface.";
 
-    public static final String OFFSET_KEY_CONVERTER_CLASS_CONFIG = "offset.key.converter";
-    public static final String OFFSET_KEY_CONVERTER_CLASS_DOC =
-            "Converter class for offset key Copycat data that implements the <code>Converter</code> interface.";
+    public static final String INTERNAL_KEY_CONVERTER_CLASS_CONFIG = "internal.key.converter";
+    public static final String INTERNAL_KEY_CONVERTER_CLASS_DOC =
+            "Converter class for internal key Copycat data that implements the <code>Converter</code> interface. Used for converting data like offsets and configs.";
 
-    public static final String OFFSET_VALUE_CONVERTER_CLASS_CONFIG = "offset.value.converter";
-    public static final String OFFSET_VALUE_CONVERTER_CLASS_DOC =
-            "Converter class for offset value Copycat data that implements the <code>Converter</code> interface.";
+    public static final String INTERNAL_VALUE_CONVERTER_CLASS_CONFIG = "internal.value.converter";
+    public static final String INTERNAL_VALUE_CONVERTER_CLASS_DOC =
+            "Converter class for offset value Copycat data that implements the <code>Converter</code> interface. Used for converting data like offsets and configs.";
 
     public static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG
             = "task.shutdown.graceful.timeout.ms";
@@ -95,10 +95,10 @@ public class WorkerConfig extends AbstractConfig {
                         Importance.HIGH, KEY_CONVERTER_CLASS_DOC)
                 .define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS,
                         Importance.HIGH, VALUE_CONVERTER_CLASS_DOC)
-                .define(OFFSET_KEY_CONVERTER_CLASS_CONFIG, Type.CLASS,
-                        Importance.HIGH, OFFSET_KEY_CONVERTER_CLASS_DOC)
-                .define(OFFSET_VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS,
-                        Importance.HIGH, OFFSET_VALUE_CONVERTER_CLASS_DOC)
+                .define(INTERNAL_KEY_CONVERTER_CLASS_CONFIG, Type.CLASS,
+                        Importance.HIGH, INTERNAL_KEY_CONVERTER_CLASS_DOC)
+                .define(INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS,
+                        Importance.HIGH, INTERNAL_VALUE_CONVERTER_CLASS_DOC)
                 .define(TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG, Type.LONG,
                         TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DEFAULT, Importance.LOW,
                         TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC)

copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/ConnectorConfig.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.common.config.ConfigDef.Importance;
 import org.apache.kafka.common.config.ConfigDef.Type;
 
+import java.util.Map;
 import java.util.Properties;
 
 /**
@@ -67,7 +68,7 @@ public ConnectorConfig() {
         this(new Properties());
     }
 
-    public ConnectorConfig(Properties props) {
+    public ConnectorConfig(Map<?, ?> props) {
         super(config, props);
     }
 }

copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Herder.java

@@ -19,7 +19,7 @@
 
 import org.apache.kafka.copycat.util.Callback;
 
-import java.util.Properties;
+import java.util.Map;
 
 /**
  * <p>
@@ -53,15 +53,24 @@ public interface Herder {
      * the leader herder if necessary.
      *
      * @param connectorProps user-specified properties for this job
-     * @param callback callback to invoke when the request completes
+     * @param callback       callback to invoke when the request completes
      */
-    void addConnector(Properties connectorProps, Callback<String> callback);
+    void addConnector(Map<String, String> connectorProps, Callback<String> callback);
 
     /**
      * Delete a connector job by name.
      *
-     * @param name name of the connector job to shutdown and delete
+     * @param name     name of the connector job to shutdown and delete
      * @param callback callback to invoke when the request completes
      */
     void deleteConnector(String name, Callback<Void> callback);
-}
+
+    /**
+     * Requests reconfiguration of the task. This should only be triggered by
+     * {@link HerderConnectorContext}.
+     *
+     * @param connName name of the connector that should be reconfigured
+     */
+    void requestTaskReconfiguration(String connName);
+
+}

copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/standalone/StandaloneConnectorContext.java → copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/HerderConnectorContext.java

@@ -15,20 +15,20 @@
  * limitations under the License.
  **/
 
-package org.apache.kafka.copycat.runtime.standalone;
+package org.apache.kafka.copycat.runtime;
 
 import org.apache.kafka.copycat.connector.ConnectorContext;
 
 /**
  * ConnectorContext for use with the StandaloneHerder, which maintains all connectors and tasks
  * in a single process.
  */
-class StandaloneConnectorContext implements ConnectorContext {
+public class HerderConnectorContext implements ConnectorContext {
 
-    private StandaloneHerder herder;
+    private Herder herder;
     private String connectorName;
 
-    public StandaloneConnectorContext(StandaloneHerder herder, String connectorName) {
+    public HerderConnectorContext(Herder herder, String connectorName) {
         this.herder = herder;
         this.connectorName = connectorName;
     }

copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Worker.java

@@ -52,8 +52,8 @@ public class Worker {
     private WorkerConfig config;
     private Converter keyConverter;
     private Converter valueConverter;
-    private Converter offsetKeyConverter;
-    private Converter offsetValueConverter;
+    private Converter internalKeyConverter;
+    private Converter internalValueConverter;
     private OffsetBackingStore offsetBackingStore;
     private HashMap<ConnectorTaskId, WorkerTask> tasks = new HashMap<>();
     private KafkaProducer<byte[], byte[]> producer;
@@ -71,10 +71,10 @@ public Worker(Time time, WorkerConfig config, OffsetBackingStore offsetBackingSt
         this.keyConverter.configure(config.originalsWithPrefix("key.converter."), true);
         this.valueConverter = config.getConfiguredInstance(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, Converter.class);
         this.valueConverter.configure(config.originalsWithPrefix("value.converter."), false);
-        this.offsetKeyConverter = config.getConfiguredInstance(WorkerConfig.OFFSET_KEY_CONVERTER_CLASS_CONFIG, Converter.class);
-        this.offsetKeyConverter.configure(config.originalsWithPrefix("offset.key.converter."), true);
-        this.offsetValueConverter = config.getConfiguredInstance(WorkerConfig.OFFSET_VALUE_CONVERTER_CLASS_CONFIG, Converter.class);
-        this.offsetValueConverter.configure(config.originalsWithPrefix("offset.value.converter."), false);
+        this.internalKeyConverter = config.getConfiguredInstance(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, Converter.class);
+        this.internalKeyConverter.configure(config.originalsWithPrefix("internal.key.converter."), true);
+        this.internalValueConverter = config.getConfiguredInstance(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, Converter.class);
+        this.internalValueConverter.configure(config.originalsWithPrefix("internal.value.converter."), false);
 
         this.offsetBackingStore = offsetBackingStore;
         this.offsetBackingStore.configure(config.originals());
@@ -157,9 +157,9 @@ public void addTask(ConnectorTaskId id, String taskClassName, Properties props)
         if (task instanceof SourceTask) {
             SourceTask sourceTask = (SourceTask) task;
             OffsetStorageReader offsetReader = new OffsetStorageReaderImpl(offsetBackingStore, id.connector(),
-                    offsetKeyConverter, offsetValueConverter);
+                    internalKeyConverter, internalValueConverter);
             OffsetStorageWriter offsetWriter = new OffsetStorageWriter(offsetBackingStore, id.connector(),
-                    offsetKeyConverter, offsetValueConverter);
+                    internalKeyConverter, internalValueConverter);
             workerTask = new WorkerSourceTask(id, sourceTask, keyConverter, valueConverter, producer,
                     offsetReader, offsetWriter, config, time);
         } else if (task instanceof SinkTask) {
@@ -201,4 +201,11 @@ private WorkerTask getTask(ConnectorTaskId id) {
         return task;
     }
 
+    public Converter getInternalKeyConverter() {
+        return internalKeyConverter;
+    }
+
+    public Converter getInternalValueConverter() {
+        return internalValueConverter;
+    }
 }

copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/ClusterConfigState.java

@@ -0,0 +1,86 @@
+/**
+ * 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.kafka.copycat.runtime.distributed;
+
+import org.apache.kafka.copycat.util.ConnectorTaskId;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * An immutable snapshot of the configuration state of connectors and tasks in a Copycat cluster.
+ */
+public class ClusterConfigState {
+    private Map<String, Integer> rootConfig;
+    private Map<String, Map<String, String>> connectorConfigs;
+    private Map<ConnectorTaskId, Map<String, String>> taskConfigs;
+
+    public ClusterConfigState(Map<String, Integer> rootConfig,
+                              Map<String, Map<String, String>> connectorConfigs,
+                              Map<ConnectorTaskId, Map<String, String>> taskConfigs) {
+        this.rootConfig = rootConfig;
+        this.connectorConfigs = connectorConfigs;
+        this.taskConfigs = taskConfigs;
+    }
+
+    /**
+     * Get a list of the connectors in this configuration
+     */
+    public Collection<String> connectors() {
+        return rootConfig.keySet();
+    }
+
+    /**
+     * Get the configuration for a connector.
+     * @param connector name of the connector
+     * @return a map containing configuration parameters
+     */
+    public Map<String, String> connectorConfig(String connector) {
+        return connectorConfigs.get(connector);
+    }
+
+    /**
+     * Get the configuration for a task.
+     * @param task id of the task
+     * @return a map containing configuration parameters
+     */
+    public Map<String, String> taskConfig(ConnectorTaskId task) {
+        return taskConfigs.get(task);
+    }
+
+    /**
+     * Get the current set of task IDs for the specified connector.
+     * @param connectorName the name of the connector to look up task configs for
+     * @return the current set of connector task IDs
+     */
+    public Collection<ConnectorTaskId> tasks(String connectorName) {
+        Integer numTasks = rootConfig.get(connectorName);
+        if (numTasks == null)
+            throw new IllegalArgumentException("Connector does not exist in current configuration.");
+
+        List<ConnectorTaskId> taskIds = new ArrayList<>();
+        for (int taskIndex = 0; taskIndex < numTasks; taskIndex++) {
+            ConnectorTaskId taskId = new ConnectorTaskId(connectorName, taskIndex);
+            taskIds.add(taskId);
+        }
+        return taskIds;
+    }
+
+}