kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject [19/26] kafka git commit: KAFKA-2774: Rename Copycat to Kafka Connect
Date Mon, 09 Nov 2015 06:11:58 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
new file mode 100644
index 0000000..85db168
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
@@ -0,0 +1,920 @@
+/**
+ * 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.connect.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.connect.connector.ConnectorContext;
+import org.apache.kafka.connect.errors.AlreadyExistsException;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.errors.NotFoundException;
+import org.apache.kafka.connect.runtime.ConnectorConfig;
+import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.HerderConnectorContext;
+import org.apache.kafka.connect.runtime.TaskConfig;
+import org.apache.kafka.connect.runtime.Worker;
+import org.apache.kafka.connect.runtime.rest.RestServer;
+import org.apache.kafka.connect.runtime.rest.entities.ConnectorInfo;
+import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
+import org.apache.kafka.connect.sink.SinkConnector;
+import org.apache.kafka.connect.storage.KafkaConfigStorage;
+import org.apache.kafka.connect.util.Callback;
+import org.apache.kafka.connect.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 ConnectProtocol.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 ConnectException("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 ConnectException("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 (ConnectException 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 ConnectException("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(ConnectProtocol.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/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/NotLeaderException.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/NotLeaderException.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/NotLeaderException.java
new file mode 100644
index 0000000..5f94b53
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/NotLeaderException.java
@@ -0,0 +1,47 @@
+/**
+ * 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.connect.runtime.distributed;
+
+import org.apache.kafka.connect.errors.ConnectException;
+
+/**
+ * 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 ConnectException {
+    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/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
new file mode 100644
index 0000000..082e235
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerCoordinator.java
@@ -0,0 +1,294 @@
+/**
+ * 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.connect.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.connect.storage.KafkaConfigStorage;
+import org.apache.kafka.connect.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 assignments
+ * to workers.
+ */
+public final class WorkerCoordinator extends AbstractCoordinator implements Closeable {
+    private static final Logger log = LoggerFactory.getLogger(WorkerCoordinator.class);
+
+    // Currently doesn't support multiple task assignment strategies, so we just fill in a default value
+    public static final String DEFAULT_SUBPROTOCOL = "default";
+
+    private final String restUrl;
+    private final KafkaConfigStorage configStorage;
+    private ConnectProtocol.Assignment assignmentSnapshot;
+    private final WorkerCoordinatorMetrics 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 WorkerCoordinatorMetrics(metrics, metricGrpPrefix, metricTags);
+        this.listener = listener;
+        this.rejoinRequested = false;
+    }
+
+    public void requestRejoin() {
+        rejoinRequested = true;
+    }
+
+    @Override
+    public String protocolType() {
+        return "connect";
+    }
+
+    @Override
+    public LinkedHashMap<String, ByteBuffer> metadata() {
+        LinkedHashMap<String, ByteBuffer> metadata = new LinkedHashMap<>();
+        configSnapshot = configStorage.snapshot();
+        ConnectProtocol.WorkerState workerState = new ConnectProtocol.WorkerState(restUrl, configSnapshot.offset());
+        metadata.put(DEFAULT_SUBPROTOCOL, ConnectProtocol.serializeMetadata(workerState));
+        return metadata;
+    }
+
+    @Override
+    protected void onJoinComplete(int generation, String memberId, String protocol, ByteBuffer memberAssignment) {
+        assignmentSnapshot = ConnectProtocol.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, ConnectProtocol.WorkerState> allConfigs = new HashMap<>();
+        for (Map.Entry<String, ByteBuffer> entry : allMemberMetadata.entrySet())
+            allConfigs.put(entry.getKey(), ConnectProtocol.deserializeMetadata(entry.getValue()));
+
+        long maxOffset = findMaxMemberConfigOffset(allConfigs);
+        Long leaderOffset = ensureLeaderConfig(maxOffset);
+        if (leaderOffset == null)
+            return fillAssignmentsAndSerialize(allConfigs.keySet(), ConnectProtocol.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, ConnectProtocol.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, ConnectProtocol.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, ConnectProtocol.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(), ConnectProtocol.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();
+            ConnectProtocol.Assignment assignment = new ConnectProtocol.Assignment(error, leaderId, leaderUrl, maxOffset, connectors, tasks);
+            log.debug("Assignment: {} -> {}", member, assignment);
+            groupAssignment.put(member, ConnectProtocol.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 WorkerCoordinatorMetrics {
+        public final Metrics metrics;
+        public final String metricGrpName;
+
+        public WorkerCoordinatorMetrics(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/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java
new file mode 100644
index 0000000..c72e3ef
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java
@@ -0,0 +1,185 @@
+/**
+ * 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.connect.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.connect.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 Connect 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 CONNECT_CLIENT_ID_SEQUENCE = new AtomicInteger(1);
+    private static final String JMX_PREFIX = "kafka.connect";
+
+    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 ? "connect-" + CONNECT_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 = "connect";
+            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("Connect 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 Connect 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 Connect group member", firstException.get());
+        else
+            log.debug("The Connect group member has stopped.");
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerRebalanceListener.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerRebalanceListener.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerRebalanceListener.java
new file mode 100644
index 0000000..40f55d2
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerRebalanceListener.java
@@ -0,0 +1,38 @@
+/**
+ * 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.connect.runtime.distributed;
+
+import org.apache.kafka.connect.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 Connect worker group. This is invoked for both successful
+     * and unsuccessful assignments.
+     */
+    void onAssigned(ConnectProtocol.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/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java
new file mode 100644
index 0000000..96346ad
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestServer.java
@@ -0,0 +1,258 @@
+/**
+ * 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.connect.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.connect.errors.ConnectException;
+import org.apache.kafka.connect.runtime.Herder;
+import org.apache.kafka.connect.runtime.WorkerConfig;
+import org.apache.kafka.connect.runtime.rest.entities.ErrorMessage;
+import org.apache.kafka.connect.runtime.rest.errors.ConnectExceptionMapper;
+import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
+import org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource;
+import org.apache.kafka.connect.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 Kafka Connect 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(ConnectExceptionMapper.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 ConnectException("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 ConnectException("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-connect");
+            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 ConnectRestException(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 ConnectRestException(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 ConnectRestException(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/417e283d/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorInfo.java
----------------------------------------------------------------------
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorInfo.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorInfo.java
new file mode 100644
index 0000000..8daae05
--- /dev/null
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/ConnectorInfo.java
@@ -0,0 +1,81 @@
+/**
+ * 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.connect.runtime.rest.entities;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.kafka.connect.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.connect.util.ConnectorTaskId> tasks) {
+        List<ConnectorTaskId> jsonTasks = new ArrayList<>();
+        for (ConnectorTaskId task : tasks)
+            jsonTasks.add(task);
+        return jsonTasks;
+    }
+}


Mime
View raw message