kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject [07/26] kafka git commit: KAFKA-2774: Rename Copycat to Kafka Connect
Date Mon, 09 Nov 2015 06:11:26 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerder.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerder.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerder.java
deleted file mode 100644
index 96de1ca..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerder.java
+++ /dev/null
@@ -1,920 +0,0 @@
-/**
- * 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.common.errors.WakeupException;
-import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.common.utils.SystemTime;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.copycat.connector.ConnectorContext;
-import org.apache.kafka.copycat.errors.AlreadyExistsException;
-import org.apache.kafka.copycat.errors.CopycatException;
-import org.apache.kafka.copycat.errors.NotFoundException;
-import org.apache.kafka.copycat.runtime.ConnectorConfig;
-import org.apache.kafka.copycat.runtime.Herder;
-import org.apache.kafka.copycat.runtime.HerderConnectorContext;
-import org.apache.kafka.copycat.runtime.TaskConfig;
-import org.apache.kafka.copycat.runtime.Worker;
-import org.apache.kafka.copycat.runtime.rest.RestServer;
-import org.apache.kafka.copycat.runtime.rest.entities.ConnectorInfo;
-import org.apache.kafka.copycat.runtime.rest.entities.TaskInfo;
-import org.apache.kafka.copycat.sink.SinkConnector;
-import org.apache.kafka.copycat.storage.KafkaConfigStorage;
-import org.apache.kafka.copycat.util.Callback;
-import org.apache.kafka.copycat.util.ConnectorTaskId;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.PriorityQueue;
-import java.util.Queue;
-import java.util.Set;
-import java.util.concurrent.Callable;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-/**
- * <p>
- *     Distributed "herder" that coordinates with other workers to spread work across multiple processes.
- * </p>
- * <p>
- *     Under the hood, this is implemented as a group managed by Kafka's group membership facilities (i.e. the generalized
- *     group/consumer coordinator). Each instance of DistributedHerder joins the group and indicates what it's current
- *     configuration state is (where it is in the configuration log). The group coordinator selects one member to take
- *     this information and assign each instance a subset of the active connectors & tasks to execute. This assignment
- *     is currently performed in a simple round-robin fashion, but this is not guaranteed -- the herder may also choose
- *     to, e.g., use a sticky assignment to avoid the usual start/stop costs associated with connectors and tasks. Once
- *     an assignment is received, the DistributedHerder simply runs its assigned connectors and tasks in a Worker.
- * </p>
- * <p>
- *     In addition to distributing work, the DistributedHerder uses the leader determined during the work assignment
- *     to select a leader for this generation of the group who is responsible for other tasks that can only be performed
- *     by a single node at a time. Most importantly, this includes writing updated configurations for connectors and tasks,
- *     (and therefore, also for creating, destroy, and scaling up/down connectors).
- * </p>
- */
-public class DistributedHerder implements Herder, Runnable {
-    private static final Logger log = LoggerFactory.getLogger(DistributedHerder.class);
-
-    private static final long RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MS = 250;
-
-    private final Worker worker;
-    private final KafkaConfigStorage configStorage;
-    private ClusterConfigState configState;
-    private final Time time;
-
-    private final int workerSyncTimeoutMs;
-    private final int workerUnsyncBackoffMs;
-
-    private final WorkerGroupMember member;
-    private final AtomicBoolean stopping;
-    private final CountDownLatch stopLatch = new CountDownLatch(1);
-
-    // Track enough information about the current membership state to be able to determine which requests via the API
-    // and the from other nodes are safe to process
-    private boolean rebalanceResolved;
-    private CopycatProtocol.Assignment assignment;
-
-    // To handle most external requests, like creating or destroying a connector, we can use a generic request where
-    // the caller specifies all the code that should be executed.
-    private final Queue<HerderRequest> requests = new PriorityQueue<>();
-    // Config updates can be collected and applied together when possible. Also, we need to take care to rebalance when
-    // needed (e.g. task reconfiguration, which requires everyone to coordinate offset commits).
-    private Set<String> connectorConfigUpdates = new HashSet<>();
-    private boolean needsReconfigRebalance;
-
-    private final ExecutorService forwardRequestExecutor;
-
-    public DistributedHerder(DistributedConfig config, Worker worker, String restUrl) {
-        this(config, worker, null, null, restUrl, new SystemTime());
-    }
-
-    // public for testing
-    public DistributedHerder(DistributedConfig config, Worker worker, KafkaConfigStorage configStorage, WorkerGroupMember member, String restUrl, Time time) {
-        this.worker = worker;
-        if (configStorage != null) {
-            // For testing. Assume configuration has already been performed
-            this.configStorage = configStorage;
-        } else {
-            this.configStorage = new KafkaConfigStorage(worker.getInternalValueConverter(), connectorConfigCallback(), taskConfigCallback());
-            this.configStorage.configure(config.originals());
-        }
-        configState = ClusterConfigState.EMPTY;
-        this.time = time;
-
-        this.workerSyncTimeoutMs = config.getInt(DistributedConfig.WORKER_SYNC_TIMEOUT_MS_CONFIG);
-        this.workerUnsyncBackoffMs = config.getInt(DistributedConfig.WORKER_UNSYNC_BACKOFF_MS_CONFIG);
-
-        this.member = member != null ? member : new WorkerGroupMember(config, restUrl, this.configStorage, rebalanceListener());
-        stopping = new AtomicBoolean(false);
-
-        rebalanceResolved = true; // If we still need to follow up after a rebalance occurred, starting up tasks
-        needsReconfigRebalance = false;
-
-        forwardRequestExecutor = Executors.newSingleThreadExecutor();
-    }
-
-    @Override
-    public void start() {
-        Thread thread = new Thread(this, "DistributedHerder");
-        thread.start();
-    }
-
-    public void run() {
-        try {
-            log.info("Herder starting");
-
-            configStorage.start();
-
-            log.info("Herder started");
-
-            while (!stopping.get()) {
-                tick();
-            }
-
-            halt();
-
-            log.info("Herder stopped");
-        } catch (Throwable t) {
-            log.error("Uncaught exception in herder work thread, exiting: ", t);
-            stopLatch.countDown();
-            System.exit(1);
-        } finally {
-            stopLatch.countDown();
-        }
-    }
-
-    // public for testing
-    public void tick() {
-        // The main loop does two primary things: 1) drive the group membership protocol, responding to rebalance events
-        // as they occur, and 2) handle external requests targeted at the leader. All the "real" work of the herder is
-        // performed in this thread, which keeps synchronization straightforward at the cost of some operations possibly
-        // blocking up this thread (especially those in callbacks due to rebalance events).
-
-        try {
-            member.ensureActive();
-            // Ensure we're in a good state in our group. If not restart and everything should be setup to rejoin
-            if (!handleRebalanceCompleted()) return;
-        } catch (WakeupException e) {
-            // May be due to a request from another thread, or might be stopping. If the latter, we need to check the
-            // flag immediately. If the former, we need to re-run the ensureActive call since we can't handle requests
-            // unless we're in the group.
-            return;
-        }
-
-        // Process any external requests
-        final long now = time.milliseconds();
-        long nextRequestTimeoutMs = Long.MAX_VALUE;
-        while (true) {
-            final HerderRequest next;
-            synchronized (this) {
-                next = requests.peek();
-                if (next == null) {
-                    break;
-                } else if (now >= next.at) {
-                    requests.poll();
-                } else {
-                    nextRequestTimeoutMs = next.at - now;
-                    break;
-                }
-            }
-
-            try {
-                next.action().call();
-                next.callback().onCompletion(null, null);
-            } catch (Throwable t) {
-                next.callback().onCompletion(t, null);
-            }
-        }
-
-        // Process any configuration updates
-        Set<String> connectorConfigUpdatesCopy = null;
-        synchronized (this) {
-            if (needsReconfigRebalance || !connectorConfigUpdates.isEmpty()) {
-                // Connector reconfigs only need local updates since there is no coordination between workers required.
-                // However, if connectors were added or removed, work needs to be rebalanced since we have more work
-                // items to distribute among workers.
-                ClusterConfigState newConfigState = configStorage.snapshot();
-                if (!newConfigState.connectors().equals(configState.connectors()))
-                    needsReconfigRebalance = true;
-                configState = newConfigState;
-                if (needsReconfigRebalance) {
-                    // Task reconfigs require a rebalance. Request the rebalance, clean out state, and then restart
-                    // this loop, which will then ensure the rebalance occurs without any other requests being
-                    // processed until it completes.
-                    member.requestRejoin();
-                    // Any connector config updates will be addressed during the rebalance too
-                    connectorConfigUpdates.clear();
-                    needsReconfigRebalance = false;
-                    return;
-                } else if (!connectorConfigUpdates.isEmpty()) {
-                    // We can't start/stop while locked since starting connectors can cause task updates that will
-                    // require writing configs, which in turn make callbacks into this class from another thread that
-                    // require acquiring a lock. This leads to deadlock. Instead, just copy the info we need and process
-                    // the updates after unlocking.
-                    connectorConfigUpdatesCopy = connectorConfigUpdates;
-                    connectorConfigUpdates = new HashSet<>();
-                }
-            }
-        }
-        if (connectorConfigUpdatesCopy != null) {
-            // If we only have connector config updates, we can just bounce the updated connectors that are
-            // currently assigned to this worker.
-            Set<String> localConnectors = assignment == null ? Collections.<String>emptySet() : new HashSet<>(assignment.connectors());
-            for (String connectorName : connectorConfigUpdatesCopy) {
-                if (!localConnectors.contains(connectorName))
-                    continue;
-                boolean remains = configState.connectors().contains(connectorName);
-                log.info("Handling connector-only config update by {} connector {}",
-                        remains ? "restarting" : "stopping", connectorName);
-                worker.stopConnector(connectorName);
-                // The update may be a deletion, so verify we actually need to restart the connector
-                if (remains)
-                    startConnector(connectorName);
-            }
-        }
-
-        // Let the group take any actions it needs to
-        try {
-            member.poll(nextRequestTimeoutMs);
-            // Ensure we're in a good state in our group. If not restart and everything should be setup to rejoin
-            if (!handleRebalanceCompleted()) return;
-        } catch (WakeupException e) { // FIXME should not be WakeupException
-            // Ignore. Just indicates we need to check the exit flag, for requested actions, etc.
-        }
-    }
-
-    // public for testing
-    public void halt() {
-        synchronized (this) {
-            // Clean up any connectors and tasks that are still running.
-            log.info("Stopping connectors and tasks that are still assigned to this worker.");
-            for (String connName : new HashSet<>(worker.connectorNames())) {
-                try {
-                    worker.stopConnector(connName);
-                } catch (Throwable t) {
-                    log.error("Failed to shut down connector " + connName, t);
-                }
-            }
-            for (ConnectorTaskId taskId : new HashSet<>(worker.taskIds())) {
-                try {
-                    worker.stopTask(taskId);
-                } catch (Throwable t) {
-                    log.error("Failed to shut down task " + taskId, t);
-                }
-            }
-
-            member.stop();
-
-            // Explicitly fail any outstanding requests so they actually get a response and get an understandable reason
-            // for their failure
-            while (!requests.isEmpty()) {
-                HerderRequest request = requests.poll();
-                request.callback().onCompletion(new CopycatException("Worker is shutting down"), null);
-            }
-
-            if (configStorage != null)
-                configStorage.stop();
-        }
-    }
-
-    @Override
-    public void stop() {
-        log.info("Herder stopping");
-
-        stopping.set(true);
-        member.wakeup();
-        while (stopLatch.getCount() > 0) {
-            try {
-                stopLatch.await();
-            } catch (InterruptedException e) {
-                // ignore, should not happen
-            }
-        }
-
-
-        forwardRequestExecutor.shutdown();
-        try {
-            if (!forwardRequestExecutor.awaitTermination(10000, TimeUnit.MILLISECONDS))
-                forwardRequestExecutor.shutdownNow();
-        } catch (InterruptedException e) {
-            // ignore
-        }
-
-        log.info("Herder stopped");
-    }
-
-    @Override
-    public synchronized void connectors(final Callback<Collection<String>> callback) {
-        log.trace("Submitting connector listing request");
-
-        addRequest(
-                new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        if (!checkConfigSynced(callback))
-                            return null;
-
-                        callback.onCompletion(null, configState.connectors());
-                        return null;
-                    }
-                },
-                forwardErrorCallback(callback)
-        );
-    }
-
-    @Override
-    public synchronized void connectorInfo(final String connName, final Callback<ConnectorInfo> callback) {
-        log.trace("Submitting connector info request {}", connName);
-
-        addRequest(
-                new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        if (!checkConfigSynced(callback))
-                            return null;
-
-                        if (!configState.connectors().contains(connName)) {
-                            callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
-                        } else {
-                            callback.onCompletion(null, new ConnectorInfo(connName, configState.connectorConfig(connName), configState.tasks(connName)));
-                        }
-                        return null;
-                    }
-                },
-                forwardErrorCallback(callback)
-        );
-    }
-
-    @Override
-    public void connectorConfig(String connName, final Callback<Map<String, String>> callback) {
-        // Subset of connectorInfo, so piggy back on that implementation
-        log.trace("Submitting connector config read request {}", connName);
-        connectorInfo(connName, new Callback<ConnectorInfo>() {
-            @Override
-            public void onCompletion(Throwable error, ConnectorInfo result) {
-                if (error != null)
-                    callback.onCompletion(error, null);
-                else
-                    callback.onCompletion(null, result.config());
-            }
-        });
-    }
-
-    @Override
-    public void putConnectorConfig(final String connName, Map<String, String> config, final boolean allowReplace,
-                                   final Callback<Created<ConnectorInfo>> callback) {
-        final Map<String, String> connConfig;
-        if (config == null) {
-            connConfig = null;
-        } else if (!config.containsKey(ConnectorConfig.NAME_CONFIG)) {
-            connConfig = new HashMap<>(config);
-            connConfig.put(ConnectorConfig.NAME_CONFIG, connName);
-        } else {
-            connConfig = config;
-        }
-
-        log.trace("Submitting connector config write request {}", connName);
-
-        addRequest(
-                new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        log.trace("Handling connector config request {}", connName);
-                        if (!isLeader()) {
-                            callback.onCompletion(new NotLeaderException("Only the leader can set connector configs.", leaderUrl()), null);
-                            return null;
-                        }
-
-                        boolean exists = configState.connectors().contains(connName);
-                        if (!allowReplace && exists) {
-                            callback.onCompletion(new AlreadyExistsException("Connector " + connName + " already exists"), null);
-                            return null;
-                        }
-
-                        if (connConfig == null && !exists) {
-                            callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
-                            return null;
-                        }
-
-                        log.trace("Submitting connector config {} {} {}", connName, allowReplace, configState.connectors());
-                        configStorage.putConnectorConfig(connName, connConfig);
-
-                        boolean created = !exists && connConfig != null;
-                        // Note that we use the updated connector config despite the fact that we don't have an updated
-                        // snapshot yet. The existing task info should still be accurate.
-                        ConnectorInfo info = connConfig == null ? null :
-                                new ConnectorInfo(connName, connConfig, configState.tasks(connName));
-                        callback.onCompletion(null, new Created<>(created, info));
-
-                        return null;
-                    }
-                },
-                forwardErrorCallback(callback)
-        );
-    }
-
-    @Override
-    public synchronized void requestTaskReconfiguration(final String connName) {
-        log.trace("Submitting connector task reconfiguration request {}", connName);
-
-        addRequest(
-                new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        reconfigureConnectorTasksWithRetry(connName);
-                        return null;
-                    }
-                },
-                new Callback<Void>() {
-                    @Override
-                    public void onCompletion(Throwable error, Void result) {
-                        log.error("Unexpected error during task reconfiguration: ", error);
-                        log.error("Task reconfiguration for {} failed unexpectedly, this connector will not be properly reconfigured unless manually triggered.", connName);
-                    }
-                }
-        );
-    }
-
-    @Override
-    public synchronized void taskConfigs(final String connName, final Callback<List<TaskInfo>> callback) {
-        log.trace("Submitting get task configuration request {}", connName);
-
-        addRequest(
-                new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        if (!checkConfigSynced(callback))
-                            return null;
-
-                        if (!configState.connectors().contains(connName)) {
-                            callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
-                        } else {
-                            List<TaskInfo> result = new ArrayList<>();
-                            for (int i = 0; i < configState.taskCount(connName); i++) {
-                                ConnectorTaskId id = new ConnectorTaskId(connName, i);
-                                result.add(new TaskInfo(id, configState.taskConfig(id)));
-                            }
-                            callback.onCompletion(null, result);
-                        }
-                        return null;
-                    }
-                },
-                forwardErrorCallback(callback)
-        );
-    }
-
-    @Override
-    public synchronized void putTaskConfigs(final String connName, final List<Map<String, String>> configs, final Callback<Void> callback) {
-        log.trace("Submitting put task configuration request {}", connName);
-
-        addRequest(
-                new Callable<Void>() {
-                    @Override
-                    public Void call() throws Exception {
-                        if (!isLeader())
-                            callback.onCompletion(new NotLeaderException("Only the leader may write task configurations.", leaderUrl()), null);
-                        else if (!configState.connectors().contains(connName))
-                            callback.onCompletion(new NotFoundException("Connector " + connName + " not found"), null);
-                        else {
-                            configStorage.putTaskConfigs(taskConfigListAsMap(connName, configs));
-                            callback.onCompletion(null, null);
-                        }
-                        return null;
-                    }
-                },
-                forwardErrorCallback(callback)
-        );
-    }
-
-
-    // Should only be called from work thread, so synchronization should not be needed
-    private boolean isLeader() {
-        return assignment != null && member.memberId().equals(assignment.leader());
-    }
-
-    /**
-     * Get the URL for the leader's REST interface, or null if we do not have the leader's URL yet.
-     */
-    private String leaderUrl() {
-        if (assignment == null)
-            return null;
-        return assignment.leaderUrl();
-    }
-
-    /**
-     * Handle post-assignment operations, either trying to resolve issues that kept assignment from completing, getting
-     * this node into sync and its work started. Since
-     *
-     * @return false if we couldn't finish
-     */
-    private boolean handleRebalanceCompleted() {
-        if (this.rebalanceResolved)
-            return true;
-
-        // We need to handle a variety of cases after a rebalance:
-        // 1. Assignment failed
-        //  1a. We are the leader for the round. We will be leader again if we rejoin now, so we need to catch up before
-        //      even attempting to. If we can't we should drop out of the group because we will block everyone from making
-        //      progress. We can backoff and try rejoining later.
-        //  1b. We are not the leader. We might need to catch up. If we're already caught up we can rejoin immediately,
-        //      otherwise, we just want to wait indefinitely to catch up and rejoin whenver we're finally ready.
-        // 2. Assignment succeeded.
-        //  2a. We are caught up on configs. Awesome! We can proceed to run our assigned work.
-        //  2b. We need to try to catch up. We can do this potentially indefinitely because if it takes to long, we'll
-        //      be kicked out of the group anyway due to lack of heartbeats.
-
-        boolean needsReadToEnd = false;
-        long syncConfigsTimeoutMs = Long.MAX_VALUE;
-        boolean needsRejoin = false;
-        if (assignment.failed()) {
-            needsRejoin = true;
-            if (isLeader()) {
-                log.warn("Join group completed, but assignment failed and we are the leader. Reading to end of config and retrying.");
-                needsReadToEnd = true;
-                syncConfigsTimeoutMs = workerSyncTimeoutMs;
-            } else if (configState.offset() < assignment.offset()) {
-                log.warn("Join group completed, but assignment failed and we lagging. Reading to end of config and retrying.");
-                needsReadToEnd = true;
-            } else {
-                log.warn("Join group completed, but assignment failed. We were up to date, so just retrying.");
-            }
-        } else {
-            if (configState.offset() < assignment.offset()) {
-                log.warn("Catching up to assignment's config offset.");
-                needsReadToEnd = true;
-            }
-        }
-
-        if (needsReadToEnd) {
-            // Force exiting this method to avoid creating any connectors/tasks and require immediate rejoining if
-            // we timed out. This should only happen if we were the leader and didn't finish quickly enough, in which
-            // case we've waited a long time and should have already left the group OR the timeout should have been
-            // very long and not having finished also indicates we've waited longer than the session timeout.
-            if (!readConfigToEnd(syncConfigsTimeoutMs))
-                needsRejoin = true;
-        }
-
-        if (needsRejoin) {
-            member.requestRejoin();
-            return false;
-        }
-
-        // Should still validate that they match since we may have gone *past* the required offset, in which case we
-        // should *not* start any tasks and rejoin
-        if (configState.offset() != assignment.offset()) {
-            log.info("Current config state offset {} does not match group assignment {}. Forcing rebalance.", configState.offset(), assignment.offset());
-            member.requestRejoin();
-            return false;
-        }
-
-        startWork();
-
-        // We only mark this as resolved once we've actually started work, which allows us to correctly track whether
-        // what work is currently active and running. If we bail early, the main tick loop + having requested rejoin
-        // guarantees we'll attempt to rejoin before executing this method again.
-        rebalanceResolved = true;
-        return true;
-    }
-
-    /**
-     * Try to read to the end of the config log within the given timeout
-     * @param timeoutMs maximum time to wait to sync to the end of the log
-     * @return true if successful, false if timed out
-     */
-    private boolean readConfigToEnd(long timeoutMs) {
-        log.info("Current config state offset {} is behind group assignment {}, reading to end of config log", configState.offset(), assignment.offset());
-        try {
-            configStorage.readToEnd().get(timeoutMs, TimeUnit.MILLISECONDS);
-            configState = configStorage.snapshot();
-            log.info("Finished reading to end of log and updated config snapshot, new config log offset: {}", configState.offset());
-            return true;
-        } catch (TimeoutException e) {
-            log.warn("Didn't reach end of config log quickly enough", e);
-            // TODO: With explicit leave group support, it would be good to explicitly leave the group *before* this
-            // backoff since it'll be longer than the session timeout
-            if (isLeader())
-                backoff(workerUnsyncBackoffMs);
-            return false;
-        } catch (InterruptedException | ExecutionException e) {
-            throw new CopycatException("Error trying to catch up after assignment", e);
-        }
-    }
-
-    private void backoff(long ms) {
-        Utils.sleep(ms);
-    }
-
-    private void startWork() {
-        // Start assigned connectors and tasks
-        log.info("Starting connectors and tasks using config offset {}", assignment.offset());
-        for (String connectorName : assignment.connectors()) {
-            try {
-                startConnector(connectorName);
-            } catch (ConfigException e) {
-                log.error("Couldn't instantiate connector " + connectorName + " because it has an invalid connector " +
-                        "configuration. This connector will not execute until reconfigured.", e);
-            }
-        }
-        for (ConnectorTaskId taskId : assignment.tasks()) {
-            try {
-                log.info("Starting task {}", taskId);
-                Map<String, String> configs = configState.taskConfig(taskId);
-                TaskConfig taskConfig = new TaskConfig(configs);
-                worker.addTask(taskId, taskConfig);
-            } catch (ConfigException e) {
-                log.error("Couldn't instantiate task " + taskId + " because it has an invalid task " +
-                        "configuration. This task will not execute until reconfigured.", e);
-            }
-        }
-        log.info("Finished starting connectors and tasks");
-    }
-
-    // Helper for starting a connector with the given name, which will extract & parse the config, generate connector
-    // context and add to the worker. This needs to be called from within the main worker thread for this herder.
-    private void startConnector(String connectorName) {
-        log.info("Starting connector {}", connectorName);
-        Map<String, String> configs = configState.connectorConfig(connectorName);
-        ConnectorConfig connConfig = new ConnectorConfig(configs);
-        String connName = connConfig.getString(ConnectorConfig.NAME_CONFIG);
-        ConnectorContext ctx = new HerderConnectorContext(DistributedHerder.this, connName);
-        worker.addConnector(connConfig, ctx);
-
-        // Immediately request configuration since this could be a brand new connector. However, also only update those
-        // task configs if they are actually different from the existing ones to avoid unnecessary updates when this is
-        // just restoring an existing connector.
-        reconfigureConnectorTasksWithRetry(connName);
-    }
-
-    private void reconfigureConnectorTasksWithRetry(final String connName) {
-        reconfigureConnector(connName, new Callback<Void>() {
-            @Override
-            public void onCompletion(Throwable error, Void result) {
-                // If we encountered an error, we don't have much choice but to just retry. If we don't, we could get
-                // stuck with a connector that thinks it has generated tasks, but wasn't actually successful and therefore
-                // never makes progress. The retry has to run through a HerderRequest since this callback could be happening
-                // from the HTTP request forwarding thread.
-                if (error != null) {
-                    log.error("Failed to reconfigure connector's tasks, retrying after backoff:", error);
-                    addRequest(RECONFIGURE_CONNECTOR_TASKS_BACKOFF_MS,
-                            new Callable<Void>() {
-                                @Override
-                                public Void call() throws Exception {
-                                    reconfigureConnectorTasksWithRetry(connName);
-                                    return null;
-                                }
-                            }, new Callback<Void>() {
-                                @Override
-                                public void onCompletion(Throwable error, Void result) {
-                                    log.error("Unexpected error during connector task reconfiguration: ", error);
-                                    log.error("Task reconfiguration for {} failed unexpectedly, this connector will not be properly reconfigured unless manually triggered.", connName);
-                                }
-                            }
-                    );
-                }
-            }
-        });
-    }
-
-    // Updates configurations for a connector by requesting them from the connector, filling in parameters provided
-    // by the system, then checks whether any configs have actually changed before submitting the new configs to storage
-    private void reconfigureConnector(final String connName, final Callback<Void> cb) {
-        try {
-            Map<String, String> configs = configState.connectorConfig(connName);
-            ConnectorConfig connConfig = new ConnectorConfig(configs);
-
-            List<String> sinkTopics = null;
-            if (SinkConnector.class.isAssignableFrom(connConfig.getClass(ConnectorConfig.CONNECTOR_CLASS_CONFIG)))
-                sinkTopics = connConfig.getList(ConnectorConfig.TOPICS_CONFIG);
-
-            final List<Map<String, String>> taskProps
-                    = worker.connectorTaskConfigs(connName, connConfig.getInt(ConnectorConfig.TASKS_MAX_CONFIG), sinkTopics);
-            boolean changed = false;
-            int currentNumTasks = configState.taskCount(connName);
-            if (taskProps.size() != currentNumTasks) {
-                log.debug("Change in connector task count from {} to {}, writing updated task configurations", currentNumTasks, taskProps.size());
-                changed = true;
-            } else {
-                int index = 0;
-                for (Map<String, String> taskConfig : taskProps) {
-                    if (!taskConfig.equals(configState.taskConfig(new ConnectorTaskId(connName, index)))) {
-                        log.debug("Change in task configurations, writing updated task configurations");
-                        changed = true;
-                        break;
-                    }
-                    index++;
-                }
-            }
-            if (changed) {
-                if (isLeader()) {
-                    configStorage.putTaskConfigs(taskConfigListAsMap(connName, taskProps));
-                    cb.onCompletion(null, null);
-                } else {
-                    // We cannot forward the request on the same thread because this reconfiguration can happen in as a
-                    // result of . If we blocked
-                    forwardRequestExecutor.submit(new Runnable() {
-                        @Override
-                        public void run() {
-                            try {
-                                String reconfigUrl = RestServer.urlJoin(leaderUrl(), "/connectors/" + connName + "/tasks");
-                                RestServer.httpRequest(reconfigUrl, "POST", taskProps, null);
-                                cb.onCompletion(null, null);
-                            } catch (CopycatException e) {
-                                log.error("Request to leader to reconfigure connector tasks failed", e);
-                                cb.onCompletion(e, null);
-                            }
-                        }
-                    });
-                }
-            }
-        } catch (Throwable t) {
-            cb.onCompletion(t, null);
-        }
-    }
-
-    // Common handling for requests that get config data. Checks if we are in sync with the current config, which allows
-    // us to answer requests directly. If we are not, handles invoking the callback with the appropriate error.
-    private boolean checkConfigSynced(Callback<?> callback) {
-        if (assignment == null || configState.offset() != assignment.offset()) {
-            if (!isLeader())
-                callback.onCompletion(new NotLeaderException("Cannot get config data because config is not in sync and this is not the leader", leaderUrl()), null);
-            else
-                callback.onCompletion(new CopycatException("Cannot get config data because this is the leader node, but it does not have the most up to date configs"), null);
-            return false;
-        }
-        return true;
-    }
-
-    private void addRequest(Callable<Void> action, Callback<Void> callback) {
-        addRequest(0, action, callback);
-    }
-
-    private void addRequest(long delayMs, Callable<Void> action, Callback<Void> callback) {
-        HerderRequest req = new HerderRequest(time.milliseconds() + delayMs, action, callback);
-        requests.add(req);
-        if (requests.peek() == req)
-            member.wakeup();
-    }
-
-    private class HerderRequest implements Comparable<HerderRequest> {
-        private final long at;
-        private final Callable<Void> action;
-        private final Callback<Void> callback;
-
-        public HerderRequest(long at, Callable<Void> action, Callback<Void> callback) {
-            this.at = at;
-            this.action = action;
-            this.callback = callback;
-        }
-
-        public Callable<Void> action() {
-            return action;
-        }
-
-        public Callback<Void> callback() {
-            return callback;
-        }
-
-        @Override
-        public int compareTo(HerderRequest o) {
-            return Long.compare(at, o.at);
-        }
-    }
-
-    private static final Callback<Void> forwardErrorCallback(final Callback<?> callback) {
-        return new Callback<Void>() {
-            @Override
-            public void onCompletion(Throwable error, Void result) {
-                if (error != null)
-                    callback.onCompletion(error, null);
-            }
-        };
-    };
-
-
-    // Config callbacks are triggered from the KafkaConfigStorage thread
-    private Callback<String> connectorConfigCallback() {
-        return new Callback<String>() {
-            @Override
-            public void onCompletion(Throwable error, String connector) {
-                log.info("Connector {} config updated", connector);
-                // Stage the update and wake up the work thread. Connector config *changes* only need the one connector
-                // to be bounced. However, this callback may also indicate a connector *addition*, which does require
-                // a rebalance, so we need to be careful about what operation we request.
-                synchronized (DistributedHerder.this) {
-                    connectorConfigUpdates.add(connector);
-                }
-                member.wakeup();
-            }
-        };
-    }
-
-    private Callback<List<ConnectorTaskId>> taskConfigCallback() {
-        return new Callback<List<ConnectorTaskId>>() {
-            @Override
-            public void onCompletion(Throwable error, List<ConnectorTaskId> tasks) {
-                log.info("Tasks {} configs updated", tasks);
-                // Stage the update and wake up the work thread. No need to record the set of tasks here because task reconfigs
-                // always need a rebalance to ensure offsets get committed.
-                // TODO: As an optimization, some task config updates could avoid a rebalance. In particular, single-task
-                // connectors clearly don't need any coordination.
-                synchronized (DistributedHerder.this) {
-                    needsReconfigRebalance = true;
-                }
-                member.wakeup();
-            }
-        };
-    }
-
-    // Rebalances are triggered internally from the group member, so these are always executed in the work thread.
-    private WorkerRebalanceListener rebalanceListener() {
-        return new WorkerRebalanceListener() {
-            @Override
-            public void onAssigned(CopycatProtocol.Assignment assignment) {
-                // This callback just logs the info and saves it. The actual response is handled in the main loop, which
-                // ensures the group member's logic for rebalancing can complete, potentially long-running steps to
-                // catch up (or backoff if we fail) not executed in a callback, and so we'll be able to invoke other
-                // group membership actions (e.g., we may need to explicitly leave the group if we cannot handle the
-                // assigned tasks).
-                log.info("Joined group and got assignment: {}", assignment);
-                synchronized (DistributedHerder.this) {
-                    DistributedHerder.this.assignment = assignment;
-                    rebalanceResolved = false;
-                }
-                // We *must* interrupt any poll() call since this could occur when the poll starts, and we might then
-                // sleep in the poll() for a long time. Forcing a wakeup ensures we'll get to process this event in the
-                // main thread.
-                member.wakeup();
-            }
-
-            @Override
-            public void onRevoked(String leader, Collection<String> connectors, Collection<ConnectorTaskId> tasks) {
-                log.info("Rebalance started");
-
-                // Note that since we don't reset the assignment, we we don't revoke leadership here. During a rebalance,
-                // it is still important to have a leader that can write configs, offsets, etc.
-
-                if (rebalanceResolved) {
-                    // TODO: Parallelize this. We should be able to request all connectors and tasks to stop, then wait on all of
-                    // them to finish
-                    // TODO: Technically we don't have to stop connectors at all until we know they've really been removed from
-                    // this worker. Instead, we can let them continue to run but buffer any update requests (which should be
-                    // rare anyway). This would avoid a steady stream of start/stop, which probably also includes lots of
-                    // unnecessary repeated connections to the source/sink system.
-                    for (String connectorName : connectors)
-                        worker.stopConnector(connectorName);
-                    // TODO: We need to at least commit task offsets, but if we could commit offsets & pause them instead of
-                    // stopping them then state could continue to be reused when the task remains on this worker. For example,
-                    // this would avoid having to close a connection and then reopen it when the task is assigned back to this
-                    // worker again.
-                    for (ConnectorTaskId taskId : tasks)
-                        worker.stopTask(taskId);
-
-                    log.info("Finished stopping tasks in preparation for rebalance");
-                } else {
-                    log.info("Wasn't unable to resume work after last rebalance, can skip stopping connectors and tasks");
-                }
-
-            }
-        };
-    }
-
-
-    private static Map<ConnectorTaskId, Map<String, String>> taskConfigListAsMap(String connName, List<Map<String, String>> configs) {
-        int index = 0;
-        Map<ConnectorTaskId, Map<String, String>> result = new HashMap<>();
-        for (Map<String, String> taskConfigMap : configs) {
-            ConnectorTaskId taskId = new ConnectorTaskId(connName, index);
-            result.put(taskId, taskConfigMap);
-            index++;
-        }
-        return result;
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/NotLeaderException.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/NotLeaderException.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/NotLeaderException.java
deleted file mode 100644
index 7e6dc67..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/NotLeaderException.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/**
- * 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.errors.CopycatException;
-
-/**
- * Indicates an operation was not permitted because it can only be performed on the leader and this worker is not currently
- * the leader.
- */
-public class NotLeaderException extends CopycatException {
-    private final String leaderUrl;
-
-    public NotLeaderException(String msg, String leaderUrl) {
-        super(msg);
-        this.leaderUrl = leaderUrl;
-    }
-
-    public NotLeaderException(String msg, String leaderUrl, Throwable throwable) {
-        super(msg, throwable);
-        this.leaderUrl = leaderUrl;
-    }
-
-    public NotLeaderException(String leaderUrl, Throwable throwable) {
-        super(throwable);
-        this.leaderUrl = leaderUrl;
-    }
-
-    public String leaderUrl() {
-        return leaderUrl;
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinator.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinator.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinator.java
deleted file mode 100644
index c748971..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinator.java
+++ /dev/null
@@ -1,293 +0,0 @@
-/**
- * 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.clients.consumer.internals.AbstractCoordinator;
-import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
-import org.apache.kafka.common.MetricName;
-import org.apache.kafka.common.metrics.Measurable;
-import org.apache.kafka.common.metrics.MetricConfig;
-import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.utils.CircularIterator;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.copycat.storage.KafkaConfigStorage;
-import org.apache.kafka.copycat.util.ConnectorTaskId;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.Closeable;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * This class manages the coordination process with the Kafka group coordinator on the broker for managing Copycat assignments to workers.
- */
-public final class WorkerCoordinator extends AbstractCoordinator implements Closeable {
-    private static final Logger log = LoggerFactory.getLogger(WorkerCoordinator.class);
-
-    // Currently Copycat doesn't support multiple task assignment strategies, so we currently just fill in a default value
-    public static final String DEFAULT_SUBPROTOCOL = "default";
-
-    private final String restUrl;
-    private final KafkaConfigStorage configStorage;
-    private CopycatProtocol.Assignment assignmentSnapshot;
-    private final CopycatWorkerCoordinatorMetrics sensors;
-    private ClusterConfigState configSnapshot;
-    private final WorkerRebalanceListener listener;
-
-    private boolean rejoinRequested;
-
-    /**
-     * Initialize the coordination manager.
-     */
-    public WorkerCoordinator(ConsumerNetworkClient client,
-                             String groupId,
-                             int sessionTimeoutMs,
-                             int heartbeatIntervalMs,
-                             Metrics metrics,
-                             String metricGrpPrefix,
-                             Map<String, String> metricTags,
-                             Time time,
-                             long requestTimeoutMs,
-                             long retryBackoffMs,
-                             String restUrl,
-                             KafkaConfigStorage configStorage,
-                             WorkerRebalanceListener listener) {
-        super(client,
-                groupId,
-                sessionTimeoutMs,
-                heartbeatIntervalMs,
-                metrics,
-                metricGrpPrefix,
-                metricTags,
-                time,
-                retryBackoffMs);
-        this.restUrl = restUrl;
-        this.configStorage = configStorage;
-        this.assignmentSnapshot = null;
-        this.sensors = new CopycatWorkerCoordinatorMetrics(metrics, metricGrpPrefix, metricTags);
-        this.listener = listener;
-        this.rejoinRequested = false;
-    }
-
-    public void requestRejoin() {
-        rejoinRequested = true;
-    }
-
-    @Override
-    public String protocolType() {
-        return "copycat";
-    }
-
-    @Override
-    public LinkedHashMap<String, ByteBuffer> metadata() {
-        LinkedHashMap<String, ByteBuffer> metadata = new LinkedHashMap<>();
-        configSnapshot = configStorage.snapshot();
-        CopycatProtocol.WorkerState workerState = new CopycatProtocol.WorkerState(restUrl, configSnapshot.offset());
-        metadata.put(DEFAULT_SUBPROTOCOL, CopycatProtocol.serializeMetadata(workerState));
-        return metadata;
-    }
-
-    @Override
-    protected void onJoinComplete(int generation, String memberId, String protocol, ByteBuffer memberAssignment) {
-        assignmentSnapshot = CopycatProtocol.deserializeAssignment(memberAssignment);
-        // At this point we always consider ourselves to be a member of the cluster, even if there was an assignment
-        // error (the leader couldn't make the assignment) or we are behind the config and cannot yet work on our assigned
-        // tasks. It's the responsibility of the code driving this process to decide how to react (e.g. trying to get
-        // up to date, try to rejoin again, leaving the group and backing off, etc.).
-        rejoinRequested = false;
-        listener.onAssigned(assignmentSnapshot);
-    }
-
-    @Override
-    protected Map<String, ByteBuffer> performAssignment(String leaderId, String protocol, Map<String, ByteBuffer> allMemberMetadata) {
-        log.debug("Performing task assignment");
-
-        Map<String, CopycatProtocol.WorkerState> allConfigs = new HashMap<>();
-        for (Map.Entry<String, ByteBuffer> entry : allMemberMetadata.entrySet())
-            allConfigs.put(entry.getKey(), CopycatProtocol.deserializeMetadata(entry.getValue()));
-
-        long maxOffset = findMaxMemberConfigOffset(allConfigs);
-        Long leaderOffset = ensureLeaderConfig(maxOffset);
-        if (leaderOffset == null)
-            return fillAssignmentsAndSerialize(allConfigs.keySet(), CopycatProtocol.Assignment.CONFIG_MISMATCH,
-                    leaderId, allConfigs.get(leaderId).url(), maxOffset,
-                    new HashMap<String, List<String>>(), new HashMap<String, List<ConnectorTaskId>>());
-        return performTaskAssignment(leaderId, leaderOffset, allConfigs);
-    }
-
-    private long findMaxMemberConfigOffset(Map<String, CopycatProtocol.WorkerState> allConfigs) {
-        // The new config offset is the maximum seen by any member. We always perform assignment using this offset,
-        // even if some members have fallen behind. The config offset used to generate the assignment is included in
-        // the response so members that have fallen behind will not use the assignment until they have caught up.
-        Long maxOffset = null;
-        for (Map.Entry<String, CopycatProtocol.WorkerState> stateEntry : allConfigs.entrySet()) {
-            long memberRootOffset = stateEntry.getValue().offset();
-            if (maxOffset == null)
-                maxOffset = memberRootOffset;
-            else
-                maxOffset = Math.max(maxOffset, memberRootOffset);
-        }
-
-        log.debug("Max config offset root: {}, local snapshot config offsets root: {}",
-                maxOffset, configSnapshot.offset());
-        return maxOffset;
-    }
-
-    private Long ensureLeaderConfig(long maxOffset) {
-        // If this leader is behind some other members, we can't do assignment
-        if (configSnapshot.offset() < maxOffset) {
-            // We might be able to take a new snapshot to catch up immediately and avoid another round of syncing here.
-            // Alternatively, if this node has already passed the maximum reported by any other member of the group, it
-            // is also safe to use this newer state.
-            ClusterConfigState updatedSnapshot = configStorage.snapshot();
-            if (updatedSnapshot.offset() < maxOffset) {
-                log.info("Was selected to perform assignments, but do not have latest config found in sync request. " +
-                        "Returning an empty configuration to trigger re-sync.");
-                return null;
-            } else {
-                configSnapshot = updatedSnapshot;
-                return configSnapshot.offset();
-            }
-        }
-
-        return maxOffset;
-    }
-
-    private Map<String, ByteBuffer> performTaskAssignment(String leaderId, long maxOffset, Map<String, CopycatProtocol.WorkerState> allConfigs) {
-        Map<String, List<String>> connectorAssignments = new HashMap<>();
-        Map<String, List<ConnectorTaskId>> taskAssignments = new HashMap<>();
-
-        // Perform round-robin task assignment
-        CircularIterator<String> memberIt = new CircularIterator<>(sorted(allConfigs.keySet()));
-        for (String connectorId : sorted(configSnapshot.connectors())) {
-            String connectorAssignedTo = memberIt.next();
-            log.trace("Assigning connector {} to {}", connectorId, connectorAssignedTo);
-            List<String> memberConnectors = connectorAssignments.get(connectorAssignedTo);
-            if (memberConnectors == null) {
-                memberConnectors = new ArrayList<>();
-                connectorAssignments.put(connectorAssignedTo, memberConnectors);
-            }
-            memberConnectors.add(connectorId);
-
-            for (ConnectorTaskId taskId : sorted(configSnapshot.tasks(connectorId))) {
-                String taskAssignedTo = memberIt.next();
-                log.trace("Assigning task {} to {}", taskId, taskAssignedTo);
-                List<ConnectorTaskId> memberTasks = taskAssignments.get(taskAssignedTo);
-                if (memberTasks == null) {
-                    memberTasks = new ArrayList<>();
-                    taskAssignments.put(taskAssignedTo, memberTasks);
-                }
-                memberTasks.add(taskId);
-            }
-        }
-
-        return fillAssignmentsAndSerialize(allConfigs.keySet(), CopycatProtocol.Assignment.NO_ERROR,
-                leaderId, allConfigs.get(leaderId).url(), maxOffset, connectorAssignments, taskAssignments);
-    }
-
-    private Map<String, ByteBuffer> fillAssignmentsAndSerialize(Collection<String> members,
-                                                                short error,
-                                                                String leaderId,
-                                                                String leaderUrl,
-                                                                long maxOffset,
-                                                                Map<String, List<String>> connectorAssignments,
-                                                                Map<String, List<ConnectorTaskId>> taskAssignments) {
-
-        Map<String, ByteBuffer> groupAssignment = new HashMap<>();
-        for (String member : members) {
-            List<String> connectors = connectorAssignments.get(member);
-            if (connectors == null)
-                connectors = Collections.emptyList();
-            List<ConnectorTaskId> tasks = taskAssignments.get(member);
-            if (tasks == null)
-                tasks = Collections.emptyList();
-            CopycatProtocol.Assignment assignment = new CopycatProtocol.Assignment(error, leaderId, leaderUrl, maxOffset, connectors, tasks);
-            log.debug("Assignment: {} -> {}", member, assignment);
-            groupAssignment.put(member, CopycatProtocol.serializeAssignment(assignment));
-        }
-        log.debug("Finished assignment");
-        return groupAssignment;
-    }
-
-    @Override
-    protected void onJoinPrepare(int generation, String memberId) {
-        log.debug("Revoking previous assignment {}", assignmentSnapshot);
-        if (assignmentSnapshot != null && !assignmentSnapshot.failed())
-            listener.onRevoked(assignmentSnapshot.leader(), assignmentSnapshot.connectors(), assignmentSnapshot.tasks());
-    }
-
-    @Override
-    public boolean needRejoin() {
-        return super.needRejoin() || (assignmentSnapshot == null || assignmentSnapshot.failed()) || rejoinRequested;
-    }
-
-    public String memberId() {
-        return this.memberId;
-    }
-
-    @Override
-    public void close() {
-        super.close();
-    }
-
-    private class CopycatWorkerCoordinatorMetrics {
-        public final Metrics metrics;
-        public final String metricGrpName;
-
-        public CopycatWorkerCoordinatorMetrics(Metrics metrics, String metricGrpPrefix, Map<String, String> tags) {
-            this.metrics = metrics;
-            this.metricGrpName = metricGrpPrefix + "-coordinator-metrics";
-
-            Measurable numConnectors = new Measurable() {
-                public double measure(MetricConfig config, long now) {
-                    return assignmentSnapshot.connectors().size();
-                }
-            };
-
-            Measurable numTasks = new Measurable() {
-                public double measure(MetricConfig config, long now) {
-                    return assignmentSnapshot.tasks().size();
-                }
-            };
-
-            metrics.addMetric(new MetricName("assigned-connectors",
-                            this.metricGrpName,
-                            "The number of connector instances currently assigned to this consumer",
-                            tags),
-                    numConnectors);
-            metrics.addMetric(new MetricName("assigned-tasks",
-                            this.metricGrpName,
-                            "The number of tasks currently assigned to this consumer",
-                            tags),
-                    numTasks);
-        }
-    }
-
-    private static <T extends Comparable<T>> List<T> sorted(Collection<T> members) {
-        List<T> res = new ArrayList<>(members);
-        Collections.sort(res);
-        return res;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerGroupMember.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerGroupMember.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerGroupMember.java
deleted file mode 100644
index 908fe59..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerGroupMember.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/**
- * 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.clients.ClientUtils;
-import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.clients.Metadata;
-import org.apache.kafka.clients.NetworkClient;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
-import org.apache.kafka.common.Cluster;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.metrics.JmxReporter;
-import org.apache.kafka.common.metrics.MetricConfig;
-import org.apache.kafka.common.metrics.Metrics;
-import org.apache.kafka.common.metrics.MetricsReporter;
-import org.apache.kafka.common.network.ChannelBuilder;
-import org.apache.kafka.common.network.Selector;
-import org.apache.kafka.common.utils.AppInfoParser;
-import org.apache.kafka.common.utils.SystemTime;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.copycat.storage.KafkaConfigStorage;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.net.InetSocketAddress;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * This class manages the coordination process with brokers for the copycat cluster group membership. It ties together
- * the Coordinator, which implements the group member protocol, with all the other pieces needed to drive the connection
- * to the group coordinator broker. This isolates all the networking to a single thread managed by this class, with
- * higher level operations in response to group membership events being handled by the herder.
- */
-public class WorkerGroupMember {
-    private static final Logger log = LoggerFactory.getLogger(WorkerGroupMember.class);
-
-    private static final AtomicInteger COPYCAT_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
-    private static final String JMX_PREFIX = "kafka.copycat";
-
-    private final Time time;
-    private final String clientId;
-    private final ConsumerNetworkClient client;
-    private final Metrics metrics;
-    private final Metadata metadata;
-    private final long retryBackoffMs;
-    private final WorkerCoordinator coordinator;
-
-    private boolean stopped = false;
-
-    public WorkerGroupMember(DistributedConfig config, String restUrl, KafkaConfigStorage configStorage, WorkerRebalanceListener listener) {
-        try {
-            this.time = new SystemTime();
-
-            MetricConfig metricConfig = new MetricConfig().samples(config.getInt(CommonClientConfigs.METRICS_NUM_SAMPLES_CONFIG))
-                    .timeWindow(config.getLong(CommonClientConfigs.METRICS_SAMPLE_WINDOW_MS_CONFIG), TimeUnit.MILLISECONDS);
-            String clientIdConfig = config.getString(CommonClientConfigs.CLIENT_ID_CONFIG);
-            clientId = clientIdConfig.length() <= 0 ? "copycat-" + COPYCAT_CLIENT_ID_SEQUENCE.getAndIncrement() : clientIdConfig;
-            List<MetricsReporter> reporters = config.getConfiguredInstances(CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG, MetricsReporter.class);
-            reporters.add(new JmxReporter(JMX_PREFIX));
-            this.metrics = new Metrics(metricConfig, reporters, time);
-            this.retryBackoffMs = config.getLong(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG);
-            this.metadata = new Metadata(retryBackoffMs, config.getLong(CommonClientConfigs.METADATA_MAX_AGE_CONFIG));
-            List<InetSocketAddress> addresses = ClientUtils.parseAndValidateAddresses(config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
-            this.metadata.update(Cluster.bootstrap(addresses), 0);
-            String metricGrpPrefix = "copycat";
-            Map<String, String> metricsTags = new LinkedHashMap<>();
-            metricsTags.put("client-id", clientId);
-            ChannelBuilder channelBuilder = ClientUtils.createChannelBuilder(config.values());
-            NetworkClient netClient = new NetworkClient(
-                    new Selector(config.getLong(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG), metrics, time, metricGrpPrefix, metricsTags, channelBuilder),
-                    this.metadata,
-                    clientId,
-                    100, // a fixed large enough value will suffice
-                    config.getLong(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG),
-                    config.getInt(CommonClientConfigs.SEND_BUFFER_CONFIG),
-                    config.getInt(CommonClientConfigs.RECEIVE_BUFFER_CONFIG),
-                    config.getInt(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG), time);
-            this.client = new ConsumerNetworkClient(netClient, metadata, time, retryBackoffMs);
-            this.coordinator = new WorkerCoordinator(this.client,
-                    config.getString(DistributedConfig.GROUP_ID_CONFIG),
-                    config.getInt(DistributedConfig.SESSION_TIMEOUT_MS_CONFIG),
-                    config.getInt(DistributedConfig.HEARTBEAT_INTERVAL_MS_CONFIG),
-                    metrics,
-                    metricGrpPrefix,
-                    metricsTags,
-                    this.time,
-                    config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
-                    retryBackoffMs,
-                    restUrl,
-                    configStorage,
-                    listener);
-
-            AppInfoParser.registerAppInfo(JMX_PREFIX, clientId);
-            log.debug("Copycat group member created");
-        } catch (Throwable t) {
-            // call close methods if internal objects are already constructed
-            // this is to prevent resource leak. see KAFKA-2121
-            stop(true);
-            // now propagate the exception
-            throw new KafkaException("Failed to construct kafka consumer", t);
-        }
-    }
-
-    public void stop() {
-        if (stopped) return;
-        stop(false);
-    }
-
-    public void ensureActive() {
-        coordinator.ensureCoordinatorKnown();
-        coordinator.ensureActiveGroup();
-    }
-
-    public void poll(long timeout) {
-        if (timeout < 0)
-            throw new IllegalArgumentException("Timeout must not be negative");
-
-        // poll for new data until the timeout expires
-        long remaining = timeout;
-        while (remaining >= 0) {
-            long start = time.milliseconds();
-            coordinator.ensureCoordinatorKnown();
-            coordinator.ensureActiveGroup();
-            client.poll(remaining);
-            remaining -= time.milliseconds() - start;
-        }
-    }
-
-    /**
-     * Interrupt any running poll() calls, causing a WakeupException to be thrown in the thread invoking that method.
-     */
-    public void wakeup() {
-        this.client.wakeup();
-    }
-
-    /**
-     * Get the member ID of this worker in the group of workers.
-     *
-     * This ID is the unique member ID automatically generated.
-     *
-     * @return the member ID
-     */
-    public String memberId() {
-        return coordinator.memberId();
-    }
-
-    public void requestRejoin() {
-        coordinator.requestRejoin();
-    }
-
-    private void stop(boolean swallowException) {
-        log.trace("Stopping the Copycat group member.");
-        AtomicReference<Throwable> firstException = new AtomicReference<Throwable>();
-        this.stopped = true;
-        ClientUtils.closeQuietly(coordinator, "coordinator", firstException);
-        ClientUtils.closeQuietly(metrics, "consumer metrics", firstException);
-        ClientUtils.closeQuietly(client, "consumer network client", firstException);
-        AppInfoParser.unregisterAppInfo(JMX_PREFIX, clientId);
-        if (firstException.get() != null && !swallowException)
-            throw new KafkaException("Failed to stop the Copycat group member", firstException.get());
-        else
-            log.debug("The Copycat group member has stopped.");
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerRebalanceListener.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerRebalanceListener.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerRebalanceListener.java
deleted file mode 100644
index c9d2ed2..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/distributed/WorkerRebalanceListener.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/**
- * 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.Collection;
-
-/**
- * Listener for rebalance events in the worker group.
- */
-public interface WorkerRebalanceListener {
-    /**
-     * Invoked when a new assignment is created by joining the Copycat worker group. This is invoked for both successful
-     * and unsuccessful assignments.
-     */
-    void onAssigned(CopycatProtocol.Assignment assignment);
-
-    /**
-     * Invoked when a rebalance operation starts, revoking ownership for the set of connectors and tasks.
-     */
-    void onRevoked(String leader, Collection<String> connectors, Collection<ConnectorTaskId> tasks);
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/RestServer.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/RestServer.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/RestServer.java
deleted file mode 100644
index 5da747d..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/RestServer.java
+++ /dev/null
@@ -1,258 +0,0 @@
-/**
- * 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.rest;
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.jaxrs.json.JacksonJsonProvider;
-import org.apache.kafka.copycat.errors.CopycatException;
-import org.apache.kafka.copycat.runtime.Herder;
-import org.apache.kafka.copycat.runtime.WorkerConfig;
-import org.apache.kafka.copycat.runtime.rest.entities.ErrorMessage;
-import org.apache.kafka.copycat.runtime.rest.errors.CopycatExceptionMapper;
-import org.apache.kafka.copycat.runtime.rest.errors.CopycatRestException;
-import org.apache.kafka.copycat.runtime.rest.resources.ConnectorsResource;
-import org.apache.kafka.copycat.runtime.rest.resources.RootResource;
-import org.eclipse.jetty.server.Connector;
-import org.eclipse.jetty.server.Handler;
-import org.eclipse.jetty.server.Server;
-import org.eclipse.jetty.server.ServerConnector;
-import org.eclipse.jetty.server.Slf4jRequestLog;
-import org.eclipse.jetty.server.handler.DefaultHandler;
-import org.eclipse.jetty.server.handler.HandlerCollection;
-import org.eclipse.jetty.server.handler.RequestLogHandler;
-import org.eclipse.jetty.server.handler.StatisticsHandler;
-import org.eclipse.jetty.servlet.ServletContextHandler;
-import org.eclipse.jetty.servlet.ServletHolder;
-import org.glassfish.jersey.server.ResourceConfig;
-import org.glassfish.jersey.servlet.ServletContainer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.UriBuilder;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.HttpURLConnection;
-import java.net.URL;
-import java.util.List;
-import java.util.Map;
-
-/**
- * Embedded server for the REST API that provides the control plane for Copycat workers.
- */
-public class RestServer {
-    private static final Logger log = LoggerFactory.getLogger(RestServer.class);
-
-    private static final long GRACEFUL_SHUTDOWN_TIMEOUT_MS = 60 * 1000;
-
-    private static final ObjectMapper JSON_SERDE = new ObjectMapper();
-
-    private final WorkerConfig config;
-    private Herder herder;
-    private Server jettyServer;
-
-    /**
-     * Create a REST server for this herder using the specified configs.
-     */
-    public RestServer(WorkerConfig config) {
-        this.config = config;
-
-        // To make the advertised port available immediately, we need to do some configuration here
-        String hostname = config.getString(WorkerConfig.REST_HOST_NAME_CONFIG);
-        Integer port = config.getInt(WorkerConfig.REST_PORT_CONFIG);
-
-        jettyServer = new Server();
-
-        ServerConnector connector = new ServerConnector(jettyServer);
-        if (hostname != null && !hostname.isEmpty())
-            connector.setHost(hostname);
-        connector.setPort(port);
-        jettyServer.setConnectors(new Connector[]{connector});
-    }
-
-    public void start(Herder herder) {
-        log.info("Starting REST server");
-
-        this.herder = herder;
-
-        ResourceConfig resourceConfig = new ResourceConfig();
-        resourceConfig.register(new JacksonJsonProvider());
-
-        resourceConfig.register(RootResource.class);
-        resourceConfig.register(new ConnectorsResource(herder));
-
-        resourceConfig.register(CopycatExceptionMapper.class);
-
-        ServletContainer servletContainer = new ServletContainer(resourceConfig);
-        ServletHolder servletHolder = new ServletHolder(servletContainer);
-
-        ServletContextHandler context = new ServletContextHandler(ServletContextHandler.SESSIONS);
-        context.setContextPath("/");
-        context.addServlet(servletHolder, "/*");
-
-        RequestLogHandler requestLogHandler = new RequestLogHandler();
-        Slf4jRequestLog requestLog = new Slf4jRequestLog();
-        requestLog.setLoggerName(RestServer.class.getCanonicalName());
-        requestLog.setLogLatency(true);
-        requestLogHandler.setRequestLog(requestLog);
-
-        HandlerCollection handlers = new HandlerCollection();
-        handlers.setHandlers(new Handler[]{context, new DefaultHandler(), requestLogHandler});
-
-        /* Needed for graceful shutdown as per `setStopTimeout` documentation */
-        StatisticsHandler statsHandler = new StatisticsHandler();
-        statsHandler.setHandler(handlers);
-        jettyServer.setHandler(statsHandler);
-        jettyServer.setStopTimeout(GRACEFUL_SHUTDOWN_TIMEOUT_MS);
-        jettyServer.setStopAtShutdown(true);
-
-        try {
-            jettyServer.start();
-        } catch (Exception e) {
-            throw new CopycatException("Unable to start REST server", e);
-        }
-
-        log.info("REST server listening at " + jettyServer.getURI() + ", advertising URL " + advertisedUrl());
-    }
-
-    public void stop() {
-        try {
-            jettyServer.stop();
-            jettyServer.join();
-        } catch (Exception e) {
-            throw new CopycatException("Unable to stop REST server", e);
-        } finally {
-            jettyServer.destroy();
-        }
-    }
-
-    /**
-     * Get the URL to advertise to other workers and clients. This uses the default connector from the embedded Jetty
-     * server, unless overrides for advertised hostname and/or port are provided via configs.
-     */
-    public String advertisedUrl() {
-        UriBuilder builder = UriBuilder.fromUri(jettyServer.getURI());
-        String advertisedHostname = config.getString(WorkerConfig.REST_ADVERTISED_HOST_NAME_CONFIG);
-        if (advertisedHostname != null && !advertisedHostname.isEmpty())
-            builder.host(advertisedHostname);
-        Integer advertisedPort = config.getInt(WorkerConfig.REST_ADVERTISED_PORT_CONFIG);
-        if (advertisedPort != null)
-            builder.port(advertisedPort);
-        else
-            builder.port(config.getInt(WorkerConfig.REST_PORT_CONFIG));
-        return builder.build().toString();
-    }
-
-
-    /**
-     * @param url               HTTP connection will be established with this url.
-     * @param method            HTTP method ("GET", "POST", "PUT", etc.)
-     * @param requestBodyData   Object to serialize as JSON and send in the request body.
-     * @param responseFormat    Expected format of the response to the HTTP request.
-     * @param <T>               The type of the deserialized response to the HTTP request.
-     * @return The deserialized response to the HTTP request, or null if no data is expected.
-     */
-    public static <T> HttpResponse<T> httpRequest(String url, String method, Object requestBodyData,
-                                    TypeReference<T> responseFormat) {
-        HttpURLConnection connection = null;
-        try {
-            String serializedBody = requestBodyData == null ? null : JSON_SERDE.writeValueAsString(requestBodyData);
-            log.debug("Sending {} with input {} to {}", method, serializedBody, url);
-
-            connection = (HttpURLConnection) new URL(url).openConnection();
-            connection.setRequestMethod(method);
-
-            connection.setRequestProperty("User-Agent", "kafka-copycat");
-            connection.setRequestProperty("Accept", "application/json");
-
-            // connection.getResponseCode() implicitly calls getInputStream, so always set to true.
-            // On the other hand, leaving this out breaks nothing.
-            connection.setDoInput(true);
-
-            connection.setUseCaches(false);
-
-            if (requestBodyData != null) {
-                connection.setRequestProperty("Content-Type", "application/json");
-                connection.setDoOutput(true);
-
-                OutputStream os = connection.getOutputStream();
-                os.write(serializedBody.getBytes());
-                os.flush();
-                os.close();
-            }
-
-            int responseCode = connection.getResponseCode();
-            if (responseCode == HttpURLConnection.HTTP_NO_CONTENT) {
-                return new HttpResponse<>(responseCode, connection.getHeaderFields(), null);
-            } else if (responseCode >= 400) {
-                InputStream es = connection.getErrorStream();
-                ErrorMessage errorMessage = JSON_SERDE.readValue(es, ErrorMessage.class);
-                es.close();
-                throw new CopycatRestException(responseCode, errorMessage.errorCode(), errorMessage.message());
-            } else if (responseCode >= 200 && responseCode < 300) {
-                InputStream is = connection.getInputStream();
-                T result = JSON_SERDE.readValue(is, responseFormat);
-                is.close();
-                return new HttpResponse<>(responseCode, connection.getHeaderFields(), result);
-            } else {
-                throw new CopycatRestException(Response.Status.INTERNAL_SERVER_ERROR,
-                        Response.Status.INTERNAL_SERVER_ERROR.getStatusCode(),
-                        "Unexpected status code when handling forwarded request: " + responseCode);
-            }
-        } catch (IOException e) {
-            log.error("IO error forwarding REST request: ", e);
-            throw new CopycatRestException(Response.Status.INTERNAL_SERVER_ERROR, "IO Error trying to forward REST request: " + e.getMessage(), e);
-        } finally {
-            if (connection != null)
-                connection.disconnect();
-        }
-    }
-
-    public static class HttpResponse<T> {
-        private int status;
-        private Map<String, List<String>> headers;
-        private T body;
-
-        public HttpResponse(int status, Map<String, List<String>> headers, T body) {
-            this.status = status;
-            this.headers = headers;
-            this.body = body;
-        }
-
-        public int status() {
-            return status;
-        }
-
-        public Map<String, List<String>> headers() {
-            return headers;
-        }
-
-        public T body() {
-            return body;
-        }
-    }
-
-    public static String urlJoin(String base, String path) {
-        if (base.endsWith("/") && path.startsWith("/"))
-            return base + path.substring(1);
-        else
-            return base + path;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/entities/ConnectorInfo.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/entities/ConnectorInfo.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/entities/ConnectorInfo.java
deleted file mode 100644
index 2b047d3..0000000
--- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/rest/entities/ConnectorInfo.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/**
- * 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.rest.entities;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import org.apache.kafka.copycat.util.ConnectorTaskId;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-
-public class ConnectorInfo {
-
-    private final String name;
-    private final Map<String, String> config;
-    private final List<ConnectorTaskId> tasks;
-
-    @JsonCreator
-    public ConnectorInfo(@JsonProperty("name") String name, @JsonProperty("config") Map<String, String> config,
-                         @JsonProperty("tasks") List<ConnectorTaskId> tasks) {
-        this.name = name;
-        this.config = config;
-        this.tasks = tasks;
-    }
-
-    @JsonProperty
-    public String name() {
-        return name;
-    }
-
-    @JsonProperty
-    public Map<String, String> config() {
-        return config;
-    }
-
-    @JsonProperty
-    public List<ConnectorTaskId> tasks() {
-        return tasks;
-    }
-
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        ConnectorInfo that = (ConnectorInfo) o;
-        return Objects.equals(name, that.name) &&
-                Objects.equals(config, that.config) &&
-                Objects.equals(tasks, that.tasks);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(name, config, tasks);
-    }
-
-
-    private static List<ConnectorTaskId> jsonTasks(Collection<org.apache.kafka.copycat.util.ConnectorTaskId> tasks) {
-        List<ConnectorTaskId> jsonTasks = new ArrayList<>();
-        for (ConnectorTaskId task : tasks)
-            jsonTasks.add(task);
-        return jsonTasks;
-    }
-}


Mime
View raw message