kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject [1/3] kafka git commit: KAFKA-2371: Add distributed support for Copycat.
Date Fri, 23 Oct 2015 23:37:40 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk 21443f214 -> 2e6177359


http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerderTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerderTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerderTest.java
index 0463b85..1213656 100644
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerderTest.java
+++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerderTest.java
@@ -17,20 +17,19 @@
 
 package org.apache.kafka.copycat.runtime.distributed;
 
-import org.apache.kafka.copycat.connector.Connector;
-import org.apache.kafka.copycat.connector.Task;
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.copycat.connector.ConnectorContext;
 import org.apache.kafka.copycat.runtime.ConnectorConfig;
-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.sink.SinkConnector;
-import org.apache.kafka.copycat.sink.SinkTask;
 import org.apache.kafka.copycat.source.SourceConnector;
 import org.apache.kafka.copycat.source.SourceTask;
 import org.apache.kafka.copycat.storage.KafkaConfigStorage;
 import org.apache.kafka.copycat.util.Callback;
 import org.apache.kafka.copycat.util.ConnectorTaskId;
-import org.apache.kafka.copycat.util.FutureCallback;
+import org.apache.kafka.copycat.util.TestFuture;
 import org.easymock.EasyMock;
+import org.easymock.IAnswer;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -39,251 +38,354 @@ import org.powermock.api.easymock.annotation.Mock;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
+import org.powermock.reflect.Whitebox;
 
-import java.util.*;
-import java.util.concurrent.TimeUnit;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeoutException;
 
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({DistributedHerder.class})
+@PrepareForTest(DistributedHerder.class)
 @PowerMockIgnore("javax.management.*")
 public class DistributedHerderTest {
-    private static final List<String> CONNECTOR_NAMES = Arrays.asList("source-test1", "source-test2", "sink-test3");
-    private static final List<String> SOURCE_CONNECTOR_NAMES = Arrays.asList("source-test1", "source-test2");
-    private static final List<String> SINK_CONNECTOR_NAMES = Arrays.asList("sink-test3");
-    private static final String TOPICS_LIST_STR = "topic1,topic2";
+    private static final Map<String, String> HERDER_CONFIG = new HashMap<>();
+    static {
+        HERDER_CONFIG.put(KafkaConfigStorage.CONFIG_TOPIC_CONFIG, "config-topic");
+        HERDER_CONFIG.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        HERDER_CONFIG.put(DistributedHerderConfig.GROUP_ID_CONFIG, "test-copycat-group");
+    }
 
-    private static final Map<String, String> CONFIG_STORAGE_CONFIG = Collections.singletonMap(KafkaConfigStorage.CONFIG_TOPIC_CONFIG, "config-topic");
+    private static final String CONN1 = "sourceA";
+    private static final String CONN2 = "sourceA";
+    private static final ConnectorTaskId TASK0 = new ConnectorTaskId(CONN1, 0);
+    private static final ConnectorTaskId TASK1 = new ConnectorTaskId(CONN1, 1);
+    private static final ConnectorTaskId TASK2 = new ConnectorTaskId(CONN1, 2);
+    private static final Integer MAX_TASKS = 3;
+    private static final Map<String, String> CONNECTOR_CONFIG = new HashMap<>();
+    static {
+        CONNECTOR_CONFIG.put(ConnectorConfig.NAME_CONFIG, "sourceA");
+        CONNECTOR_CONFIG.put(ConnectorConfig.TASKS_MAX_CONFIG, MAX_TASKS.toString());
+        CONNECTOR_CONFIG.put(ConnectorConfig.TOPICS_CONFIG, "foo,bar");
+        CONNECTOR_CONFIG.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, BogusSourceConnector.class.getName());
+    }
+    private static final Map<String, String> TASK_CONFIG = new HashMap<>();
+    static {
+        TASK_CONFIG.put(TaskConfig.TASK_CLASS_CONFIG, BogusSourceTask.class.getName());
+    }
+    private static final HashMap<ConnectorTaskId, Map<String, String>> TASK_CONFIGS = new HashMap<>();
+    static {
+        TASK_CONFIGS.put(TASK0, TASK_CONFIG);
+        TASK_CONFIGS.put(TASK1, TASK_CONFIG);
+        TASK_CONFIGS.put(TASK2, TASK_CONFIG);
+    }
+    private static final ClusterConfigState SNAPSHOT = new ClusterConfigState(1, Collections.singletonMap(CONN1, 3),
+            Collections.singletonMap(CONN1, CONNECTOR_CONFIG), TASK_CONFIGS, Collections.<String>emptySet());
 
     @Mock private KafkaConfigStorage configStorage;
+    @Mock private WorkerGroupMember member;
     private DistributedHerder herder;
     @Mock private Worker worker;
     @Mock private Callback<String> createCallback;
+    @Mock private Callback<Void> destroyCallback;
 
-    private Map<String, Map<String, String>> connectorProps;
-    private Map<String, Class<? extends Connector>> connectorClasses;
-    private Map<String, Class<? extends Task>> connectorTaskClasses;
-    private Map<String, Connector> connectors;
-    private Properties taskProps;
+    private Callback<String> connectorConfigCallback;
+    private Callback<List<ConnectorTaskId>> taskConfigCallback;
+    private WorkerRebalanceListener rebalanceListener;
 
     @Before
-    public void setUp() {
+    public void setUp() throws Exception {
         worker = PowerMock.createMock(Worker.class);
-        herder = new DistributedHerder(worker, configStorage);
-
-        connectorProps = new HashMap<>();
-        connectorClasses = new HashMap<>();
-        connectorTaskClasses = new HashMap<>();
-        connectors = new HashMap<>();
-        for (String connectorName : CONNECTOR_NAMES) {
-            Class<? extends Connector> connectorClass = connectorName.contains("source") ? BogusSourceConnector.class : BogusSinkConnector.class;
-            Class<? extends Task> taskClass = connectorName.contains("source") ? BogusSourceTask.class : BogusSinkTask.class;
-            Connector connector = connectorName.contains("source") ? PowerMock.createMock(BogusSourceConnector.class) : PowerMock.createMock(BogusSinkConnector.class);
-
-            Map<String, String> props = new HashMap<>();
-            props.put(ConnectorConfig.NAME_CONFIG, connectorName);
-            props.put(SinkConnector.TOPICS_CONFIG, TOPICS_LIST_STR);
-            props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, connectorClass.getName());
-
-            connectorProps.put(connectorName, props);
-            connectorClasses.put(connectorName, connectorClass);
-            connectorTaskClasses.put(connectorName, taskClass);
-            connectors.put(connectorName, connector);
-        }
-
-        PowerMock.mockStatic(DistributedHerder.class);
-
-        // These can be anything since connectors can pass along whatever they want.
-        taskProps = new Properties();
-        taskProps.setProperty("foo", "bar");
+
+        herder = PowerMock.createPartialMock(DistributedHerder.class, new String[]{"backoff"},
+                worker, HERDER_CONFIG, configStorage, member);
+        connectorConfigCallback = Whitebox.invokeMethod(herder, "connectorConfigCallback");
+        taskConfigCallback = Whitebox.invokeMethod(herder, "taskConfigCallback");
+        rebalanceListener = Whitebox.invokeMethod(herder, "rebalanceListener");
     }
 
     @Test
-    public void testCreateSourceConnector() throws Exception {
-        String connectorName = SOURCE_CONNECTOR_NAMES.get(0);
+    public void testJoinAssignment() {
+        // Join group and get assignment
+        EasyMock.expect(member.memberId()).andStubReturn("member");
+        expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1));
+        expectPostRebalanceCatchup(SNAPSHOT);
+        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
+        PowerMock.expectLastCall();
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
+        worker.addTask(EasyMock.eq(TASK1), EasyMock.<TaskConfig>anyObject());
+        PowerMock.expectLastCall();
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
 
-        expectConfigStorageConfigureStart();
-        expectEmptyRestore();
-        expectAdd(connectorName);
         PowerMock.replayAll();
 
-        herder.configure(CONFIG_STORAGE_CONFIG);
-        herder.start();
-        herder.addConnector(connectorProps.get(connectorName), createCallback);
+        herder.tick();
 
         PowerMock.verifyAll();
     }
 
     @Test
-    public void testCreateSinkConnector() throws Exception {
-        String connectorName = SINK_CONNECTOR_NAMES.get(0);
+    public void testHaltCleansUpWorker() {
+        EasyMock.expect(worker.connectorNames()).andReturn(Collections.singleton(CONN1));
+        worker.stopConnector(CONN1);
+        PowerMock.expectLastCall();
+        EasyMock.expect(worker.taskIds()).andReturn(Collections.singleton(TASK1));
+        worker.stopTask(TASK1);
+        PowerMock.expectLastCall();
+        member.stop();
+        PowerMock.expectLastCall();
+        configStorage.stop();
+        PowerMock.expectLastCall();
 
-        expectConfigStorageConfigureStart();
-        expectEmptyRestore();
-        expectAdd(connectorName);
         PowerMock.replayAll();
 
-        herder.configure(CONFIG_STORAGE_CONFIG);
-        herder.start();
-        herder.addConnector(connectorProps.get(connectorName), createCallback);
+        herder.halt();
 
         PowerMock.verifyAll();
     }
 
     @Test
-    public void testDestroyConnector() throws Exception {
-        String connectorName = SOURCE_CONNECTOR_NAMES.get(0);
-
-        expectConfigStorageConfigureStart();
-        expectEmptyRestore();
-        expectAdd(connectorName);
-        expectDestroy(connectorName);
-        PowerMock.replayAll();
+    public void testCreateConnector() throws Exception {
+        EasyMock.expect(member.memberId()).andStubReturn("leader");
+        expectRebalance(1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
+        expectPostRebalanceCatchup(SNAPSHOT);
 
-        herder.configure(CONFIG_STORAGE_CONFIG);
-        herder.start();
-        herder.addConnector(connectorProps.get(connectorName), createCallback);
+        member.wakeup();
+        PowerMock.expectLastCall();
+        configStorage.putConnectorConfig(CONN1, CONNECTOR_CONFIG);
+        PowerMock.expectLastCall();
+        createCallback.onCompletion(null, CONN1);
+        PowerMock.expectLastCall();
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
+        // No immediate action besides this -- change will be picked up via the config log
 
-        FutureCallback<Void> futureCb = new FutureCallback<>(new Callback<Void>() {
-            @Override
-            public void onCompletion(Throwable error, Void result) {
+        PowerMock.replayAll();
 
-            }
-        });
-        herder.deleteConnector(CONNECTOR_NAMES.get(0), futureCb);
-        futureCb.get(1000L, TimeUnit.MILLISECONDS);
+        herder.addConnector(CONNECTOR_CONFIG, createCallback);
+        herder.tick();
 
         PowerMock.verifyAll();
     }
 
     @Test
-    public void testCreateAndStop() throws Exception {
-        String connectorName = SOURCE_CONNECTOR_NAMES.get(0);
+    public void testDestroyConnector() throws Exception {
+        EasyMock.expect(member.memberId()).andStubReturn("leader");
+        // Start with one connector
+        expectRebalance(1, Arrays.asList(CONN1), Collections.<ConnectorTaskId>emptyList());
+        expectPostRebalanceCatchup(SNAPSHOT);
+        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
+        PowerMock.expectLastCall();
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
+
+        // And delete the connector
+        member.wakeup();
+        PowerMock.expectLastCall();
+        configStorage.putConnectorConfig(CONN1, null);
+        PowerMock.expectLastCall();
+        destroyCallback.onCompletion(null, null);
+        PowerMock.expectLastCall();
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
+        // No immediate action besides this -- change will be picked up via the config log
 
-        expectConfigStorageConfigureStart();
-        expectEmptyRestore();
-        expectAdd(connectorName);
         PowerMock.replayAll();
 
-        herder.configure(CONFIG_STORAGE_CONFIG);
-        herder.start();
-        herder.addConnector(connectorProps.get(connectorName), createCallback);
+        herder.deleteConnector(CONN1, destroyCallback);
+        herder.tick();
 
         PowerMock.verifyAll();
     }
 
     @Test
-    public void testRestoreAndStop() throws Exception {
-        String restoreConnectorName1 = SOURCE_CONNECTOR_NAMES.get(0);
-        String restoreConnectorName2 = SINK_CONNECTOR_NAMES.get(0);
-        String additionalConnectorName = SOURCE_CONNECTOR_NAMES.get(1);
-
-        expectConfigStorageConfigureStart();
-        expectRestore(Arrays.asList(restoreConnectorName1, restoreConnectorName2));
-        expectAdd(additionalConnectorName);
-        // Stopping the herder should correctly stop all restored and new connectors
-        expectStop(restoreConnectorName1);
-        expectStop(restoreConnectorName2);
-        expectStop(additionalConnectorName);
-        configStorage.stop();
+    public void testConnectorConfigAdded() {
+        // If a connector was added, we need to rebalance
+        EasyMock.expect(member.memberId()).andStubReturn("member");
+
+        // join, no configs so no need to catch up on config topic
+        expectRebalance(-1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
+
+        // apply config
+        member.wakeup();
+        member.ensureActive();
+        PowerMock.expectLastCall();
+        // Checks for config updates and starts rebalance
+        EasyMock.expect(configStorage.snapshot()).andReturn(SNAPSHOT);
+        member.requestRejoin();
+        PowerMock.expectLastCall();
+        // Performs rebalance and gets new assignment
+        expectRebalance(Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(),
+                CopycatProtocol.Assignment.NO_ERROR, 1, Arrays.asList(CONN1), Collections.<ConnectorTaskId>emptyList());
+        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
+        PowerMock.expectLastCall();
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
+        member.poll(EasyMock.anyInt());
         PowerMock.expectLastCall();
 
         PowerMock.replayAll();
 
-        herder.configure(CONFIG_STORAGE_CONFIG);
-        herder.start();
-        herder.addConnector(connectorProps.get(additionalConnectorName), createCallback);
-        herder.stop();
+        herder.tick(); // join
+        connectorConfigCallback.onCompletion(null, CONN1); // read updated config
+        herder.tick(); // apply config
+        herder.tick(); // do rebalance
 
         PowerMock.verifyAll();
     }
 
-    private void expectConfigStorageConfigureStart() {
-        configStorage.configure(CONFIG_STORAGE_CONFIG);
+    @Test
+    public void testConnectorConfigUpdate() {
+        // Connector config can be applied without any rebalance
+
+        EasyMock.expect(member.memberId()).andStubReturn("member");
+        EasyMock.expect(worker.connectorNames()).andStubReturn(Collections.singleton(CONN1));
+
+        // join
+        expectRebalance(1, Arrays.asList(CONN1), Collections.<ConnectorTaskId>emptyList());
+        expectPostRebalanceCatchup(SNAPSHOT);
+        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
         PowerMock.expectLastCall();
-        configStorage.start();
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
+        member.poll(EasyMock.anyInt());
         PowerMock.expectLastCall();
-    }
 
-    private void expectAdd(String connectorName) throws Exception {
-        configStorage.putConnectorConfig(connectorName, connectorProps.get(connectorName));
+        // apply config
+        member.wakeup();
+        member.ensureActive();
+        PowerMock.expectLastCall();
+        EasyMock.expect(configStorage.snapshot()).andReturn(SNAPSHOT); // for this test, it doesn't matter if we use the same config snapshot
+        worker.stopConnector(CONN1);
+        PowerMock.expectLastCall();
+        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
+        PowerMock.expectLastCall();
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
+        member.poll(EasyMock.anyInt());
         PowerMock.expectLastCall();
-        expectInstantiateConnector(connectorName, true);
-    }
 
-    private void expectEmptyRestore() throws Exception {
-        expectRestore(Collections.<String>emptyList());
-    }
+        PowerMock.replayAll();
+
+        herder.tick(); // join
+        connectorConfigCallback.onCompletion(null, CONN1); // read updated config
+        herder.tick(); // apply config
 
-    private void expectRestore(List<String> connectorNames) throws Exception {
-        Map<String, Integer> rootConfig = new HashMap<>();
-        Map<String, Map<String, String>> connectorConfigs = new HashMap<>();
-        for (String connName : connectorNames) {
-            rootConfig.put(connName, 0);
-            connectorConfigs.put(connName, connectorProps.get(connName));
-        }
-        EasyMock.expect(configStorage.snapshot())
-                .andReturn(new ClusterConfigState(1, rootConfig, connectorConfigs, Collections.EMPTY_MAP, Collections.EMPTY_SET));
-
-        // Restore never uses a callback
-        for (String connectorName : connectorNames)
-            expectInstantiateConnector(connectorName, false);
+        PowerMock.verifyAll();
     }
 
-    private void expectInstantiateConnector(String connectorName, boolean expectCallback) throws Exception {
-        PowerMock.expectPrivate(DistributedHerder.class, "instantiateConnector", connectorClasses.get(connectorName).getName())
-                .andReturn(connectors.get(connectorName));
-        if (expectCallback) {
-            createCallback.onCompletion(null, connectorName);
-            PowerMock.expectLastCall();
-        }
+    @Test
+    public void testTaskConfigAdded() {
+        // Task config always requires rebalance
+        EasyMock.expect(member.memberId()).andStubReturn("member");
+
+        // join
+        expectRebalance(-1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
 
-        Connector connector = connectors.get(connectorName);
-        connector.initialize(EasyMock.anyObject(HerderConnectorContext.class));
+        // apply config
+        member.wakeup();
+        member.ensureActive();
+        PowerMock.expectLastCall();
+        // Checks for config updates and starts rebalance
+        EasyMock.expect(configStorage.snapshot()).andReturn(SNAPSHOT);
+        member.requestRejoin();
         PowerMock.expectLastCall();
-        connector.start(new Properties());
+        // Performs rebalance and gets new assignment
+        expectRebalance(Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(),
+                CopycatProtocol.Assignment.NO_ERROR, 1, Collections.<String>emptyList(), Arrays.asList(TASK0));
+        worker.addTask(EasyMock.eq(TASK0), EasyMock.<TaskConfig>anyObject());
         PowerMock.expectLastCall();
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
+
+        PowerMock.replayAll();
 
-        // Just return the connector properties for the individual task we generate by default
-        EasyMock.<Class<? extends Task>>expect(connector.taskClass()).andReturn(connectorTaskClasses.get(connectorName));
+        herder.tick(); // join
+        taskConfigCallback.onCompletion(null, Arrays.asList(TASK0, TASK1, TASK2)); // read updated config
+        herder.tick(); // apply config
+        herder.tick(); // do rebalance
 
-        EasyMock.expect(connector.taskConfigs(ConnectorConfig.TASKS_MAX_DEFAULT))
-                .andReturn(Arrays.asList(taskProps));
-        // And we should instantiate the tasks. For a sink task, we should see added properties for
-        // the input topic partitions
-        Properties generatedTaskProps = new Properties();
-        generatedTaskProps.putAll(taskProps);
-        if (connectorName.contains("sink"))
-            generatedTaskProps.setProperty(SinkTask.TOPICS_CONFIG, TOPICS_LIST_STR);
-        ConnectorTaskId taskId = new ConnectorTaskId(connectorName, 0);
-        worker.addTask(taskId, connectorTaskClasses.get(connectorName).getName(), generatedTaskProps);
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testJoinLeaderCatchUpFails() throws Exception {
+        // Join group and as leader fail to do assignment
+        EasyMock.expect(member.memberId()).andStubReturn("leader");
+        expectRebalance(Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(),
+                CopycatProtocol.Assignment.CONFIG_MISMATCH, 1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
+        // Reading to end of log times out
+        TestFuture<Void> readToEndFuture = new TestFuture<>();
+        readToEndFuture.resolveOnGet(new TimeoutException());
+        EasyMock.expect(configStorage.readToEnd()).andReturn(readToEndFuture);
+        PowerMock.expectPrivate(herder, "backoff", DistributedHerderConfig.WORKER_UNSYNC_BACKOFF_MS_DEFAULT);
+        member.requestRejoin();
+
+        // After backoff, restart the process and this time succeed
+        expectRebalance(1, Arrays.asList(CONN1), Arrays.asList(TASK1));
+        expectPostRebalanceCatchup(SNAPSHOT);
+
+        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
         PowerMock.expectLastCall();
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
+        worker.addTask(EasyMock.eq(TASK1), EasyMock.<TaskConfig>anyObject());
+        PowerMock.expectLastCall();
+        member.poll(EasyMock.anyInt());
+        PowerMock.expectLastCall();
+
+        PowerMock.replayAll();
+
+        herder.tick();
+        herder.tick();
+
+        PowerMock.verifyAll();
     }
 
-    private void expectStop(String connectorName) {
-        worker.stopTask(new ConnectorTaskId(connectorName, 0));
-        EasyMock.expectLastCall();
-        Connector connector = connectors.get(connectorName);
-        connector.stop();
-        EasyMock.expectLastCall();
+    @Test
+    public void testInconsistentConfigs() throws Exception {
+        // FIXME: if we have inconsistent configs, we need to request forced reconfig + write of the connector's task configs
+        // This requires inter-worker communication, so needs the REST API
     }
 
-    private void expectDestroy(String connectorName) {
-        expectStop(connectorName);
-        configStorage.putConnectorConfig(connectorName, null);
-        PowerMock.expectLastCall();
+
+    private void expectRebalance(final long offset, final List<String> assignedConnectors, final List<ConnectorTaskId> assignedTasks) {
+        expectRebalance(null, null, CopycatProtocol.Assignment.NO_ERROR, offset, assignedConnectors, assignedTasks);
     }
 
-    // We need to use a real class here due to some issue with mocking java.lang.Class
-    private abstract class BogusSourceConnector extends SourceConnector {
+    // Handles common initial part of rebalance callback. Does not handle instantiation of connectors and tasks.
+    private void expectRebalance(final Collection<String> revokedConnectors, final List<ConnectorTaskId> revokedTasks,
+                                 final short error, final long offset, final List<String> assignedConnectors, final List<ConnectorTaskId> assignedTasks) {
+        member.ensureActive();
+        PowerMock.expectLastCall().andAnswer(new IAnswer<Object>() {
+            @Override
+            public Object answer() throws Throwable {
+                if (revokedConnectors != null)
+                    rebalanceListener.onRevoked("leader", revokedConnectors, revokedTasks);
+                CopycatProtocol.Assignment assignment = new CopycatProtocol.Assignment(
+                        error, "leader", offset, assignedConnectors, assignedTasks);
+                rebalanceListener.onAssigned(assignment);
+                return null;
+            }
+        });
     }
 
-    private abstract class BogusSourceTask extends SourceTask {
+    private void expectPostRebalanceCatchup(final ClusterConfigState readToEndSnapshot) {
+        TestFuture<Void> readToEndFuture = new TestFuture<>();
+        readToEndFuture.resolveOnGet((Void) null);
+        EasyMock.expect(configStorage.readToEnd()).andReturn(readToEndFuture);
+        EasyMock.expect(configStorage.snapshot()).andReturn(readToEndSnapshot);
     }
 
-    private abstract class BogusSinkConnector extends SinkConnector {
+
+    // We need to use a real class here due to some issue with mocking java.lang.Class
+    private abstract class BogusSourceConnector extends SourceConnector {
     }
 
-    private abstract class BogusSinkTask extends SourceTask {
+    private abstract class BogusSourceTask extends SourceTask {
     }
 
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinatorTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinatorTest.java
new file mode 100644
index 0000000..30c76a2
--- /dev/null
+++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/WorkerCoordinatorTest.java
@@ -0,0 +1,436 @@
+/**
+ * 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.ClientRequest;
+import org.apache.kafka.clients.Metadata;
+import org.apache.kafka.clients.MockClient;
+import org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient;
+import org.apache.kafka.common.Cluster;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.Struct;
+import org.apache.kafka.common.requests.GroupMetadataResponse;
+import org.apache.kafka.common.requests.JoinGroupResponse;
+import org.apache.kafka.common.requests.SyncGroupRequest;
+import org.apache.kafka.common.requests.SyncGroupResponse;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.copycat.storage.KafkaConfigStorage;
+import org.apache.kafka.copycat.util.ConnectorTaskId;
+import org.apache.kafka.test.TestUtils;
+import org.easymock.EasyMock;
+import org.easymock.Mock;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.powermock.api.easymock.PowerMock;
+import org.powermock.reflect.Whitebox;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+public class WorkerCoordinatorTest {
+
+    private String connectorId = "connector";
+    private String connectorId2 = "connector2";
+    private ConnectorTaskId taskId0 = new ConnectorTaskId(connectorId, 0);
+    private ConnectorTaskId taskId1 = new ConnectorTaskId(connectorId, 1);
+    private ConnectorTaskId taskId2 = new ConnectorTaskId(connectorId2, 0);
+
+    private String groupId = "test-group";
+    private int sessionTimeoutMs = 10;
+    private int heartbeatIntervalMs = 2;
+    private long retryBackoffMs = 100;
+    private long requestTimeoutMs = 5000;
+    private MockTime time;
+    private MockClient client;
+    private Cluster cluster = TestUtils.singletonCluster("topic", 1);
+    private Node node = cluster.nodes().get(0);
+    private Metadata metadata;
+    private Metrics metrics;
+    private Map<String, String> metricTags = new LinkedHashMap<>();
+    private ConsumerNetworkClient consumerClient;
+    private MockRebalanceListener rebalanceListener;
+    @Mock private KafkaConfigStorage configStorage;
+    private WorkerCoordinator coordinator;
+
+    private ClusterConfigState configState1;
+    private ClusterConfigState configState2;
+
+    @Before
+    public void setup() {
+        this.time = new MockTime();
+        this.client = new MockClient(time);
+        this.metadata = new Metadata(0, Long.MAX_VALUE);
+        this.metadata.update(cluster, time.milliseconds());
+        this.consumerClient = new ConsumerNetworkClient(client, metadata, time, 100);
+        this.metrics = new Metrics(time);
+        this.rebalanceListener = new MockRebalanceListener();
+        this.configStorage = PowerMock.createMock(KafkaConfigStorage.class);
+
+        client.setNode(node);
+
+        this.coordinator = new WorkerCoordinator(consumerClient,
+                groupId,
+                sessionTimeoutMs,
+                heartbeatIntervalMs,
+                metrics,
+                "consumer" + groupId,
+                metricTags,
+                time,
+                requestTimeoutMs,
+                retryBackoffMs,
+                configStorage,
+                rebalanceListener);
+
+        configState1 = new ClusterConfigState(
+                1L, Collections.singletonMap(connectorId, 1),
+                Collections.singletonMap(connectorId, (Map<String, String>) new HashMap<String, String>()),
+                Collections.singletonMap(taskId0, (Map<String, String>) new HashMap<String, String>()),
+                Collections.<String>emptySet()
+        );
+        Map<String, Integer> configState2ConnectorTaskCounts = new HashMap<>();
+        configState2ConnectorTaskCounts.put(connectorId, 2);
+        configState2ConnectorTaskCounts.put(connectorId2, 1);
+        Map<String, Map<String, String>> configState2ConnectorConfigs = new HashMap<>();
+        configState2ConnectorConfigs.put(connectorId, new HashMap<String, String>());
+        configState2ConnectorConfigs.put(connectorId2, new HashMap<String, String>());
+        Map<ConnectorTaskId, Map<String, String>> configState2TaskConfigs = new HashMap<>();
+        configState2TaskConfigs.put(taskId0, new HashMap<String, String>());
+        configState2TaskConfigs.put(taskId1, new HashMap<String, String>());
+        configState2TaskConfigs.put(taskId2, new HashMap<String, String>());
+        configState2 = new ClusterConfigState(
+                2L, configState2ConnectorTaskCounts,
+                configState2ConnectorConfigs,
+                configState2TaskConfigs,
+                Collections.<String>emptySet()
+        );
+    }
+
+    @After
+    public void teardown() {
+        this.metrics.close();
+    }
+
+    // We only test functionality unique to WorkerCoordinator. Most functionality is already well tested via the tests
+    // that cover AbstractCoordinator & ConsumerCoordinator.
+
+    @Test
+    public void testMetadata() {
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+
+        PowerMock.replayAll();
+
+        LinkedHashMap<String, ByteBuffer> serialized = coordinator.metadata();
+        assertEquals(1, serialized.size());
+        CopycatProtocol.ConfigState state = CopycatProtocol.deserializeMetadata(serialized.get(WorkerCoordinator.DEFAULT_SUBPROTOCOL));
+        assertEquals(1, state.offset());
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testNormalJoinGroupLeader() {
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+
+        PowerMock.replayAll();
+
+        final String consumerId = "leader";
+
+        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        coordinator.ensureCoordinatorKnown();
+
+        // normal join group
+        Map<String, Long> memberConfigOffsets = new HashMap<>();
+        memberConfigOffsets.put("leader", 1L);
+        memberConfigOffsets.put("member", 1L);
+        client.prepareResponse(joinGroupLeaderResponse(1, consumerId, memberConfigOffsets, Errors.NONE.code()));
+        client.prepareResponse(new MockClient.RequestMatcher() {
+            @Override
+            public boolean matches(ClientRequest request) {
+                SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
+                return sync.memberId().equals(consumerId) &&
+                        sync.generationId() == 1 &&
+                        sync.groupAssignment().containsKey(consumerId);
+            }
+        }, syncGroupResponse(CopycatProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.singletonList(connectorId),
+                Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
+        coordinator.ensureActiveGroup();
+
+        assertFalse(coordinator.needRejoin());
+        assertEquals(0, rebalanceListener.revokedCount);
+        assertEquals(1, rebalanceListener.assignedCount);
+        assertFalse(rebalanceListener.assignment.failed());
+        assertEquals(1L, rebalanceListener.assignment.offset());
+        assertEquals("leader", rebalanceListener.assignment.leader());
+        assertEquals(Collections.singletonList(connectorId), rebalanceListener.assignment.connectors());
+        assertEquals(Collections.emptyList(), rebalanceListener.assignment.tasks());
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testNormalJoinGroupFollower() {
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+
+        PowerMock.replayAll();
+
+        final String memberId = "member";
+
+        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        coordinator.ensureCoordinatorKnown();
+
+        // normal join group
+        client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
+        client.prepareResponse(new MockClient.RequestMatcher() {
+            @Override
+            public boolean matches(ClientRequest request) {
+                SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
+                return sync.memberId().equals(memberId) &&
+                        sync.generationId() == 1 &&
+                        sync.groupAssignment().isEmpty();
+            }
+        }, syncGroupResponse(CopycatProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.<String>emptyList(),
+                Collections.singletonList(taskId0), Errors.NONE.code()));
+        coordinator.ensureActiveGroup();
+
+        assertFalse(coordinator.needRejoin());
+        assertEquals(0, rebalanceListener.revokedCount);
+        assertEquals(1, rebalanceListener.assignedCount);
+        assertFalse(rebalanceListener.assignment.failed());
+        assertEquals(1L, rebalanceListener.assignment.offset());
+        assertEquals(Collections.emptyList(), rebalanceListener.assignment.connectors());
+        assertEquals(Collections.singletonList(taskId0), rebalanceListener.assignment.tasks());
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testJoinLeaderCannotAssign() {
+        // If the selected leader can't get up to the maximum offset, it will fail to assign and we should immediately
+        // need to retry the join.
+
+        // When the first round fails, we'll take an updated config snapshot
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState2);
+
+        PowerMock.replayAll();
+
+        final String memberId = "member";
+
+        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        coordinator.ensureCoordinatorKnown();
+
+        // config mismatch results in assignment error
+        client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
+        MockClient.RequestMatcher matcher = new MockClient.RequestMatcher() {
+            @Override
+            public boolean matches(ClientRequest request) {
+                SyncGroupRequest sync = new SyncGroupRequest(request.request().body());
+                return sync.memberId().equals(memberId) &&
+                        sync.generationId() == 1 &&
+                        sync.groupAssignment().isEmpty();
+            }
+        };
+        client.prepareResponse(matcher, syncGroupResponse(CopycatProtocol.Assignment.CONFIG_MISMATCH, "leader", 10L,
+                Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
+        client.prepareResponse(joinGroupFollowerResponse(1, memberId, "leader", Errors.NONE.code()));
+        client.prepareResponse(matcher, syncGroupResponse(CopycatProtocol.Assignment.NO_ERROR, "leader", 1L,
+                Collections.<String>emptyList(), Collections.singletonList(taskId0), Errors.NONE.code()));
+        coordinator.ensureActiveGroup();
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testRejoinGroup() {
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+
+        PowerMock.replayAll();
+
+        client.prepareResponse(groupMetadataResponse(node, Errors.NONE.code()));
+        coordinator.ensureCoordinatorKnown();
+
+        // join the group once
+        client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
+        client.prepareResponse(syncGroupResponse(CopycatProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.<String>emptyList(),
+                Collections.singletonList(taskId0), Errors.NONE.code()));
+        coordinator.ensureActiveGroup();
+
+        assertEquals(0, rebalanceListener.revokedCount);
+        assertEquals(1, rebalanceListener.assignedCount);
+        assertFalse(rebalanceListener.assignment.failed());
+        assertEquals(1L, rebalanceListener.assignment.offset());
+        assertEquals(Collections.emptyList(), rebalanceListener.assignment.connectors());
+        assertEquals(Collections.singletonList(taskId0), rebalanceListener.assignment.tasks());
+
+        // and join the group again
+        coordinator.requestRejoin();
+        client.prepareResponse(joinGroupFollowerResponse(1, "consumer", "leader", Errors.NONE.code()));
+        client.prepareResponse(syncGroupResponse(CopycatProtocol.Assignment.NO_ERROR, "leader", 1L, Collections.singletonList(connectorId),
+                Collections.<ConnectorTaskId>emptyList(), Errors.NONE.code()));
+        coordinator.ensureActiveGroup();
+
+        assertEquals(1, rebalanceListener.revokedCount);
+        assertEquals(Collections.emptyList(), rebalanceListener.revokedConnectors);
+        assertEquals(Collections.singletonList(taskId0), rebalanceListener.revokedTasks);
+        assertEquals(2, rebalanceListener.assignedCount);
+        assertFalse(rebalanceListener.assignment.failed());
+        assertEquals(1L, rebalanceListener.assignment.offset());
+        assertEquals(Collections.singletonList(connectorId), rebalanceListener.assignment.connectors());
+        assertEquals(Collections.emptyList(), rebalanceListener.assignment.tasks());
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testLeaderDoSync1() throws Exception {
+        // Since all the protocol responses are mocked, the other tests validate doSync runs, but don't validate its
+        // output. So we test it directly here.
+
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState1);
+
+        PowerMock.replayAll();
+
+        // Prime the current configuration state
+        coordinator.metadata();
+
+        Map<String, ByteBuffer> configs = new HashMap<>();
+        // Mark everyone as in sync with configState1
+        configs.put("leader", CopycatProtocol.serializeMetadata(new CopycatProtocol.ConfigState(1L)));
+        configs.put("member", CopycatProtocol.serializeMetadata(new CopycatProtocol.ConfigState(1L)));
+        Map<String, ByteBuffer> result = Whitebox.invokeMethod(coordinator, "doSync", "leader", WorkerCoordinator.DEFAULT_SUBPROTOCOL, configs);
+
+        // configState1 has 1 connector, 1 task
+        CopycatProtocol.Assignment leaderAssignment = CopycatProtocol.deserializeAssignment(result.get("leader"));
+        assertEquals(false, leaderAssignment.failed());
+        assertEquals("leader", leaderAssignment.leader());
+        assertEquals(1, leaderAssignment.offset());
+        assertEquals(Collections.singletonList(connectorId), leaderAssignment.connectors());
+        assertEquals(Collections.emptyList(), leaderAssignment.tasks());
+
+        CopycatProtocol.Assignment memberAssignment = CopycatProtocol.deserializeAssignment(result.get("member"));
+        assertEquals(false, memberAssignment.failed());
+        assertEquals("leader", memberAssignment.leader());
+        assertEquals(1, memberAssignment.offset());
+        assertEquals(Collections.emptyList(), memberAssignment.connectors());
+        assertEquals(Collections.singletonList(taskId0), memberAssignment.tasks());
+
+        PowerMock.verifyAll();
+    }
+
+    @Test
+    public void testLeaderDoSync2() throws Exception {
+        // Since all the protocol responses are mocked, the other tests validate doSync runs, but don't validate its
+        // output. So we test it directly here.
+
+        EasyMock.expect(configStorage.snapshot()).andReturn(configState2);
+
+        PowerMock.replayAll();
+
+        // Prime the current configuration state
+        coordinator.metadata();
+
+        Map<String, ByteBuffer> configs = new HashMap<>();
+        // Mark everyone as in sync with configState1
+        configs.put("leader", CopycatProtocol.serializeMetadata(new CopycatProtocol.ConfigState(1L)));
+        configs.put("member", CopycatProtocol.serializeMetadata(new CopycatProtocol.ConfigState(1L)));
+        Map<String, ByteBuffer> result = Whitebox.invokeMethod(coordinator, "doSync", "leader", WorkerCoordinator.DEFAULT_SUBPROTOCOL, configs);
+
+        // configState2 has 2 connector, 3 tasks and should trigger round robin assignment
+        CopycatProtocol.Assignment leaderAssignment = CopycatProtocol.deserializeAssignment(result.get("leader"));
+        assertEquals(false, leaderAssignment.failed());
+        assertEquals("leader", leaderAssignment.leader());
+        assertEquals(1, leaderAssignment.offset());
+        assertEquals(Collections.singletonList(connectorId), leaderAssignment.connectors());
+        assertEquals(Arrays.asList(taskId1, taskId2), leaderAssignment.tasks());
+
+        CopycatProtocol.Assignment memberAssignment = CopycatProtocol.deserializeAssignment(result.get("member"));
+        assertEquals(false, memberAssignment.failed());
+        assertEquals("leader", memberAssignment.leader());
+        assertEquals(1, memberAssignment.offset());
+        assertEquals(Collections.singletonList(connectorId2), memberAssignment.connectors());
+        assertEquals(Collections.singletonList(taskId0), memberAssignment.tasks());
+
+        PowerMock.verifyAll();
+    }
+
+
+    private Struct groupMetadataResponse(Node node, short error) {
+        GroupMetadataResponse response = new GroupMetadataResponse(error, node);
+        return response.toStruct();
+    }
+
+    private Struct joinGroupLeaderResponse(int generationId, String memberId,
+                                           Map<String, Long> configOffsets, short error) {
+        Map<String, ByteBuffer> metadata = new HashMap<>();
+        for (Map.Entry<String, Long> configStateEntry : configOffsets.entrySet()) {
+            ByteBuffer buf = CopycatProtocol.serializeMetadata(new CopycatProtocol.ConfigState(configStateEntry.getValue()));
+            metadata.put(configStateEntry.getKey(), buf);
+        }
+        return new JoinGroupResponse(error, generationId, WorkerCoordinator.DEFAULT_SUBPROTOCOL, memberId, memberId, metadata).toStruct();
+    }
+
+    private Struct joinGroupFollowerResponse(int generationId, String memberId, String leaderId, short error) {
+        return new JoinGroupResponse(error, generationId, WorkerCoordinator.DEFAULT_SUBPROTOCOL, memberId, leaderId,
+                Collections.<String, ByteBuffer>emptyMap()).toStruct();
+    }
+
+    private Struct syncGroupResponse(short assignmentError, String leader, long configOffset, List<String> connectorIds,
+                                     List<ConnectorTaskId> taskIds, short error) {
+        CopycatProtocol.Assignment assignment = new CopycatProtocol.Assignment(assignmentError, leader, configOffset, connectorIds, taskIds);
+        ByteBuffer buf = CopycatProtocol.serializeAssignment(assignment);
+        return new SyncGroupResponse(error, buf).toStruct();
+    }
+
+
+    private static class MockRebalanceListener implements WorkerRebalanceListener {
+        public CopycatProtocol.Assignment assignment = null;
+
+        public String revokedLeader;
+        public Collection<String> revokedConnectors;
+        public Collection<ConnectorTaskId> revokedTasks;
+
+        public int revokedCount = 0;
+        public int assignedCount = 0;
+
+        @Override
+        public void onAssigned(CopycatProtocol.Assignment assignment) {
+            this.assignment = assignment;
+            assignedCount++;
+        }
+
+        @Override
+        public void onRevoked(String leader, Collection<String> connectors, Collection<ConnectorTaskId> tasks) {
+            this.revokedLeader = leader;
+            this.revokedConnectors = connectors;
+            this.revokedTasks = tasks;
+            revokedCount++;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/standalone/StandaloneHerderTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/standalone/StandaloneHerderTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/standalone/StandaloneHerderTest.java
index 606b94d..b395fc7 100644
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/standalone/StandaloneHerderTest.java
+++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/standalone/StandaloneHerderTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.copycat.connector.Connector;
 import org.apache.kafka.copycat.connector.Task;
 import org.apache.kafka.copycat.runtime.ConnectorConfig;
 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.sink.SinkConnector;
 import org.apache.kafka.copycat.sink.SinkTask;
@@ -35,22 +36,21 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.powermock.api.easymock.PowerMock;
 import org.powermock.api.easymock.annotation.Mock;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
-import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({StandaloneHerder.class})
-@PowerMockIgnore("javax.management.*")
 public class StandaloneHerderTest {
     private static final String CONNECTOR_NAME = "test";
+    private static final List<String> TOPICS_LIST = Arrays.asList("topic1", "topic2");
     private static final String TOPICS_LIST_STR = "topic1,topic2";
+    private static final int DEFAULT_MAX_TASKS = 1;
 
     private StandaloneHerder herder;
     @Mock protected Worker worker;
@@ -58,7 +58,7 @@ public class StandaloneHerderTest {
     @Mock protected Callback<String> createCallback;
 
     private Map<String, String> connectorProps;
-    private Properties taskProps;
+    private Map<String, String> taskProps;
 
     @Before
     public void setup() {
@@ -68,11 +68,10 @@ public class StandaloneHerderTest {
         connectorProps = new HashMap<>();
         connectorProps.put(ConnectorConfig.NAME_CONFIG, CONNECTOR_NAME);
         connectorProps.put(SinkConnector.TOPICS_CONFIG, TOPICS_LIST_STR);
-        PowerMock.mockStatic(StandaloneHerder.class);
 
         // These can be anything since connectors can pass along whatever they want.
-        taskProps = new Properties();
-        taskProps.setProperty("foo", "bar");
+        taskProps = new HashMap<>();
+        taskProps.put("foo", "bar");
     }
 
     @Test
@@ -121,7 +120,9 @@ public class StandaloneHerderTest {
     public void testCreateAndStop() throws Exception {
         connector = PowerMock.createMock(BogusSourceConnector.class);
         expectAdd(BogusSourceConnector.class, BogusSourceTask.class, false);
+        // herder.stop() should stop any running connectors and tasks even if destroyConnector was not invoked
         expectStop();
+
         PowerMock.replayAll();
 
         herder.addConnector(connectorProps, createCallback);
@@ -135,36 +136,30 @@ public class StandaloneHerderTest {
                            boolean sink) throws Exception {
         connectorProps.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, connClass.getName());
 
-        PowerMock.expectPrivate(StandaloneHerder.class, "instantiateConnector", connClass.getName())
-                .andReturn(connector);
-
-        createCallback.onCompletion(null, CONNECTOR_NAME);
+        worker.addConnector(EasyMock.eq(new ConnectorConfig(connectorProps)), EasyMock.anyObject(HerderConnectorContext.class));
         PowerMock.expectLastCall();
 
-        connector.initialize(EasyMock.anyObject(HerderConnectorContext.class));
-        PowerMock.expectLastCall();
-        connector.start(new Properties());
+        createCallback.onCompletion(null, CONNECTOR_NAME);
         PowerMock.expectLastCall();
 
-        // Just return the connector properties for the individual task we generate by default
-        EasyMock.<Class<? extends Task>>expect(connector.taskClass()).andReturn(taskClass);
-
-        EasyMock.expect(connector.taskConfigs(ConnectorConfig.TASKS_MAX_DEFAULT))
-                .andReturn(Arrays.asList(taskProps));
         // And we should instantiate the tasks. For a sink task, we should see added properties for
         // the input topic partitions
-        Properties generatedTaskProps = new Properties();
+        Map<String, String> generatedTaskProps = new HashMap<>();
         generatedTaskProps.putAll(taskProps);
+        generatedTaskProps.put(TaskConfig.TASK_CLASS_CONFIG, taskClass.getName());
         if (sink)
-            generatedTaskProps.setProperty(SinkTask.TOPICS_CONFIG, TOPICS_LIST_STR);
-        worker.addTask(new ConnectorTaskId(CONNECTOR_NAME, 0), taskClass.getName(), generatedTaskProps);
+            generatedTaskProps.put(SinkTask.TOPICS_CONFIG, TOPICS_LIST_STR);
+        EasyMock.expect(worker.reconfigureConnectorTasks(CONNECTOR_NAME, DEFAULT_MAX_TASKS, TOPICS_LIST))
+                .andReturn(Collections.singletonMap(new ConnectorTaskId(CONNECTOR_NAME, 0), generatedTaskProps));
+
+        worker.addTask(new ConnectorTaskId(CONNECTOR_NAME, 0), new TaskConfig(generatedTaskProps));
         PowerMock.expectLastCall();
     }
 
     private void expectStop() {
         worker.stopTask(new ConnectorTaskId(CONNECTOR_NAME, 0));
         EasyMock.expectLastCall();
-        connector.stop();
+        worker.stopConnector(CONNECTOR_NAME);
         EasyMock.expectLastCall();
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaConfigStorageTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaConfigStorageTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaConfigStorageTest.java
index b02b752..cf9f8aa 100644
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaConfigStorageTest.java
+++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaConfigStorageTest.java
@@ -169,6 +169,12 @@ public class KafkaConfigStorageTest {
         connectorReconfiguredCallback.onCompletion(null, CONNECTOR_IDS.get(1));
         EasyMock.expectLastCall();
 
+        // Config deletion
+        expectConvertWriteAndRead(
+                CONNECTOR_CONFIG_KEYS.get(1), KafkaConfigStorage.CONNECTOR_CONFIGURATION_V0, null, null, null);
+        connectorReconfiguredCallback.onCompletion(null, CONNECTOR_IDS.get(1));
+        EasyMock.expectLastCall();
+
         expectStop();
 
         PowerMock.replayAll();
@@ -185,17 +191,24 @@ public class KafkaConfigStorageTest {
         // Writing should block until it is written and read back from Kafka
         configStorage.putConnectorConfig(CONNECTOR_IDS.get(0), SAMPLE_CONFIGS.get(0));
         configState = configStorage.snapshot();
-        assertEquals(0, configState.offset());
+        assertEquals(1, configState.offset());
         assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
         assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1)));
 
         // Second should also block and all configs should still be available
         configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), SAMPLE_CONFIGS.get(1));
         configState = configStorage.snapshot();
-        assertEquals(1, configState.offset());
+        assertEquals(2, configState.offset());
         assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
         assertEquals(SAMPLE_CONFIGS.get(1), configState.connectorConfig(CONNECTOR_IDS.get(1)));
 
+        // Deletion should remove the second one we added
+        configStorage.putConnectorConfig(CONNECTOR_IDS.get(1), null);
+        configState = configStorage.snapshot();
+        assertEquals(3, configState.offset());
+        assertEquals(SAMPLE_CONFIGS.get(0), configState.connectorConfig(CONNECTOR_IDS.get(0)));
+        assertNull(configState.connectorConfig(CONNECTOR_IDS.get(1)));
+
         configStorage.stop();
 
         PowerMock.verifyAll();
@@ -255,13 +268,13 @@ public class KafkaConfigStorageTest {
 
         // Validate root config by listing all connectors and tasks
         configState = configStorage.snapshot();
-        assertEquals(2, configState.offset());
+        assertEquals(3, configState.offset());
         String connectorName = CONNECTOR_IDS.get(0);
         assertEquals(Arrays.asList(connectorName), new ArrayList<>(configState.connectors()));
-        assertEquals(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1)), configState.tasks(connectorName));
+        assertEquals(new HashSet<>(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1))), configState.tasks(connectorName));
         assertEquals(SAMPLE_CONFIGS.get(0), configState.taskConfig(TASK_IDS.get(0)));
         assertEquals(SAMPLE_CONFIGS.get(1), configState.taskConfig(TASK_IDS.get(1)));
-        assertEquals(new HashSet<>(Collections.EMPTY_LIST), configState.inconsistentConnectors());
+        assertEquals(Collections.EMPTY_SET, configState.inconsistentConnectors());
 
         configStorage.stop();
 
@@ -306,16 +319,16 @@ public class KafkaConfigStorageTest {
 
         // Should see a single connector and its config should be the last one seen anywhere in the log
         ClusterConfigState configState = configStorage.snapshot();
-        assertEquals(6, configState.offset()); // Should always be last read, even if uncommitted
+        assertEquals(7, configState.offset()); // Should always be next to be read, even if uncommitted
         assertEquals(Arrays.asList(CONNECTOR_IDS.get(0)), new ArrayList<>(configState.connectors()));
         // CONNECTOR_CONFIG_STRUCTS[2] -> SAMPLE_CONFIGS[2]
         assertEquals(SAMPLE_CONFIGS.get(2), configState.connectorConfig(CONNECTOR_IDS.get(0)));
         // Should see 2 tasks for that connector. Only config updates before the root key update should be reflected
-        assertEquals(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1)), configState.tasks(CONNECTOR_IDS.get(0)));
+        assertEquals(new HashSet<>(Arrays.asList(TASK_IDS.get(0), TASK_IDS.get(1))), configState.tasks(CONNECTOR_IDS.get(0)));
         // Both TASK_CONFIG_STRUCTS[0] -> SAMPLE_CONFIGS[0]
         assertEquals(SAMPLE_CONFIGS.get(0), configState.taskConfig(TASK_IDS.get(0)));
         assertEquals(SAMPLE_CONFIGS.get(0), configState.taskConfig(TASK_IDS.get(1)));
-        assertEquals(new HashSet<>(Collections.EMPTY_LIST), configState.inconsistentConnectors());
+        assertEquals(Collections.EMPTY_SET, configState.inconsistentConnectors());
 
         configStorage.stop();
 
@@ -374,10 +387,10 @@ public class KafkaConfigStorageTest {
         configStorage.start();
         // After reading the log, it should have been in an inconsistent state
         ClusterConfigState configState = configStorage.snapshot();
-        assertEquals(5, configState.offset()); // Should always be last read, not last committed
+        assertEquals(6, configState.offset()); // Should always be next to be read, not last committed
         assertEquals(Arrays.asList(CONNECTOR_IDS.get(0)), new ArrayList<>(configState.connectors()));
         // Inconsistent data should leave us with no tasks listed for the connector and an entry in the inconsistent list
-        assertEquals(Collections.EMPTY_LIST, configState.tasks(CONNECTOR_IDS.get(0)));
+        assertEquals(Collections.EMPTY_SET, configState.tasks(CONNECTOR_IDS.get(0)));
         // Both TASK_CONFIG_STRUCTS[0] -> SAMPLE_CONFIGS[0]
         assertNull(configState.taskConfig(TASK_IDS.get(0)));
         assertNull(configState.taskConfig(TASK_IDS.get(1)));
@@ -398,11 +411,11 @@ public class KafkaConfigStorageTest {
         configState = configStorage.snapshot();
         // This is only two more ahead of the last one because multiple calls fail, and so their configs are not written
         // to the topic. Only the last call with 1 task config + 1 commit actually gets written.
-        assertEquals(7, configState.offset());
+        assertEquals(8, configState.offset());
         assertEquals(Arrays.asList(CONNECTOR_IDS.get(0)), new ArrayList<>(configState.connectors()));
-        assertEquals(Arrays.asList(TASK_IDS.get(0)), configState.tasks(CONNECTOR_IDS.get(0)));
+        assertEquals(new HashSet<>(Arrays.asList(TASK_IDS.get(0))), configState.tasks(CONNECTOR_IDS.get(0)));
         assertEquals(SAMPLE_CONFIGS.get(0), configState.taskConfig(TASK_IDS.get(0)));
-        assertEquals(new HashSet<>(Collections.EMPTY_LIST), configState.inconsistentConnectors());
+        assertEquals(Collections.EMPTY_SET, configState.inconsistentConnectors());
 
         configStorage.stop();
 
@@ -446,17 +459,19 @@ public class KafkaConfigStorageTest {
     private void expectConvertWriteRead(final String configKey, final Schema valueSchema, final byte[] serialized,
                                         final String dataFieldName, final Object dataFieldValue) {
         final Capture<Struct> capturedRecord = EasyMock.newCapture();
-        EasyMock.expect(converter.fromCopycatData(EasyMock.eq(TOPIC), EasyMock.eq(valueSchema), EasyMock.capture(capturedRecord)))
-                .andReturn(serialized);
+        if (serialized != null)
+            EasyMock.expect(converter.fromCopycatData(EasyMock.eq(TOPIC), EasyMock.eq(valueSchema), EasyMock.capture(capturedRecord)))
+                    .andReturn(serialized);
         storeLog.send(EasyMock.eq(configKey), EasyMock.aryEq(serialized));
         PowerMock.expectLastCall();
         EasyMock.expect(converter.toCopycatData(EasyMock.eq(TOPIC), EasyMock.aryEq(serialized)))
                 .andAnswer(new IAnswer<SchemaAndValue>() {
                     @Override
                     public SchemaAndValue answer() throws Throwable {
-                        assertEquals(dataFieldValue, capturedRecord.getValue().get(dataFieldName));
+                        if (dataFieldName != null)
+                            assertEquals(dataFieldValue, capturedRecord.getValue().get(dataFieldName));
                         // Note null schema because default settings for internal serialization are schema-less
-                        return new SchemaAndValue(null, structToMap(capturedRecord.getValue()));
+                        return new SchemaAndValue(null, serialized == null ? null : structToMap(capturedRecord.getValue()));
                     }
                 });
     }

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java
index c5978ec..55e24c8 100644
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java
+++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java
@@ -102,8 +102,14 @@ public class TestFuture<T> implements Future<T> {
             }
         }
 
-        if (exception != null)
-            throw new ExecutionException(exception);
+        if (exception != null) {
+            if (exception instanceof TimeoutException)
+                throw (TimeoutException) exception;
+            else if (exception instanceof InterruptedException)
+                throw (InterruptedException) exception;
+            else
+                throw new ExecutionException(exception);
+        }
         return result;
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/tests/kafkatest/services/copycat.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/services/copycat.py b/tests/kafkatest/services/copycat.py
index 4e2ab40..45ef330 100644
--- a/tests/kafkatest/services/copycat.py
+++ b/tests/kafkatest/services/copycat.py
@@ -39,6 +39,16 @@ class CopycatServiceBase(Service):
         except:
             return []
 
+    def set_configs(self, config_template, connector_config_templates):
+        """
+        Set configurations for the worker and the connector to run on
+        it. These are not provided in the constructor because the worker
+        config generally needs access to ZK/Kafka services to
+        create the configuration.
+        """
+        self.config_template = config_template
+        self.connector_config_templates = connector_config_templates
+
     def stop_node(self, node, clean_shutdown=True):
         pids = self.pids(node)
         sig = signal.SIGTERM if clean_shutdown else signal.SIGKILL
@@ -51,7 +61,7 @@ class CopycatServiceBase(Service):
         node.account.ssh("rm -f /mnt/copycat.pid", allow_fail=False)
 
     def restart(self):
-        # We don't want to do any clean up here, just restart the process
+        # We don't want to do any clean up here, just restart the process.
         for node in self.nodes:
             self.stop_node(node)
             self.start_node(node)
@@ -62,8 +72,11 @@ class CopycatServiceBase(Service):
                              (self.__class__.__name__, node.account))
         for pid in self.pids(node):
             node.account.signal(pid, signal.SIGKILL, allow_fail=False)
-        node.account.ssh("rm -rf /mnt/copycat.pid /mnt/copycat.log /mnt/copycat.properties /mnt/copycat-connector.properties " + " ".join(self.files), allow_fail=False)
 
+        node.account.ssh("rm -rf /mnt/copycat.pid /mnt/copycat.log /mnt/copycat.properties  " + " ".join(self.config_filenames() + self.files), allow_fail=False)
+
+    def config_filenames(self):
+        return ["/mnt/copycat-connector-" + str(idx) + ".properties" for idx, template in enumerate(self.connector_config_templates)]
 
 class CopycatStandaloneService(CopycatServiceBase):
     """Runs Copycat in standalone mode."""
@@ -71,16 +84,6 @@ class CopycatStandaloneService(CopycatServiceBase):
     def __init__(self, context, kafka, files):
         super(CopycatStandaloneService, self).__init__(context, 1, kafka, files)
 
-    def set_configs(self, config_template, connector_config_template):
-        """
-        Set configurations for the worker and the connector to run on
-        it. These are not provided in the constructor because the worker
-        config generally needs access to ZK/Kafka services to
-        create the configuration.
-        """
-        self.config_template = config_template
-        self.connector_config_template = connector_config_template
-
     # For convenience since this service only makes sense with a single node
     @property
     def node(self):
@@ -88,12 +91,17 @@ class CopycatStandaloneService(CopycatServiceBase):
 
     def start_node(self, node):
         node.account.create_file("/mnt/copycat.properties", self.config_template)
-        node.account.create_file("/mnt/copycat-connector.properties", self.connector_config_template)
+        remote_connector_configs = []
+        for idx, template in enumerate(self.connector_config_templates):
+            target_file = "/mnt/copycat-connector-" + str(idx) + ".properties"
+            node.account.create_file(target_file, template)
+            remote_connector_configs.append(target_file)
 
         self.logger.info("Starting Copycat standalone process")
         with node.account.monitor_log("/mnt/copycat.log") as monitor:
-            node.account.ssh("/opt/kafka/bin/copycat-standalone.sh /mnt/copycat.properties /mnt/copycat-connector.properties " +
-                             "1>> /mnt/copycat.log 2>> /mnt/copycat.log & echo $! > /mnt/copycat.pid")
+            node.account.ssh("/opt/kafka/bin/copycat-standalone.sh /mnt/copycat.properties " +
+                             " ".join(remote_connector_configs) +
+                             " 1>> /mnt/copycat.log 2>> /mnt/copycat.log & echo $! > /mnt/copycat.pid")
             monitor.wait_until('Copycat started', timeout_sec=10, err_msg="Never saw message indicating Copycat finished startup")
 
         if len(self.pids(node)) == 0:
@@ -108,27 +116,28 @@ class CopycatDistributedService(CopycatServiceBase):
         super(CopycatDistributedService, self).__init__(context, num_nodes, kafka, files)
         self.offsets_topic = offsets_topic
         self.configs_topic = configs_topic
-
-    def set_configs(self, config_template, connector_config_template):
-        """
-        Set configurations for the worker and the connector to run on
-        it. These are not provided in the constructor because the worker
-        config generally needs access to ZK/Kafka services to
-        create the configuration.
-        """
-        self.config_template = config_template
-        self.connector_config_template = connector_config_template
+        self.first_start = True
 
     def start_node(self, node):
         node.account.create_file("/mnt/copycat.properties", self.config_template)
-        node.account.create_file("/mnt/copycat-connector.properties", self.connector_config_template)
+        remote_connector_configs = []
+        for idx, template in enumerate(self.connector_config_templates):
+            target_file = "/mnt/copycat-connector-" + str(idx) + ".properties"
+            node.account.create_file(target_file, template)
+            remote_connector_configs.append(target_file)
 
-        self.logger.info("Starting Copycat standalone process")
+        self.logger.info("Starting Copycat distributed process")
         with node.account.monitor_log("/mnt/copycat.log") as monitor:
-            node.account.ssh("/opt/kafka/bin/copycat-distributed.sh /mnt/copycat.properties /mnt/copycat-connector.properties " +
-                             "1>> /mnt/copycat.log 2>> /mnt/copycat.log & echo $! > /mnt/copycat.pid")
+            cmd = "/opt/kafka/bin/copycat-distributed.sh /mnt/copycat.properties "
+            # Only submit connectors on the first node so they don't get submitted multiple times. Also only submit them
+            # the first time the node is started so
+            if self.first_start and node == self.nodes[0]:
+                cmd += " ".join(remote_connector_configs)
+            cmd += " 1>> /mnt/copycat.log 2>> /mnt/copycat.log & echo $! > /mnt/copycat.pid"
+            node.account.ssh(cmd)
             monitor.wait_until('Copycat started', timeout_sec=10, err_msg="Never saw message indicating Copycat finished startup")
 
         if len(self.pids(node)) == 0:
             raise RuntimeError("No process ids recorded")
 
+        self.first_start = False

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/tests/kafkatest/tests/copycat_distributed_test.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/tests/copycat_distributed_test.py b/tests/kafkatest/tests/copycat_distributed_test.py
index 9d00334..57965e5 100644
--- a/tests/kafkatest/tests/copycat_distributed_test.py
+++ b/tests/kafkatest/tests/copycat_distributed_test.py
@@ -31,10 +31,12 @@ class CopycatDistributedFileTest(KafkaTest):
     OFFSETS_TOPIC = "copycat-offsets"
     CONFIG_TOPIC = "copycat-configs"
 
-    FIRST_INPUT_LISTS = [["foo", "bar", "baz"], ["foo2", "bar2", "baz2"]]
-    FIRST_INPUTS = ["\n".join(input_list) + "\n" for input_list in FIRST_INPUT_LISTS]
-    SECOND_INPUT_LISTS = [["razz", "ma", "tazz"], ["razz2", "ma2", "tazz2"]]
-    SECOND_INPUTS = ["\n".join(input_list) + "\n" for input_list in SECOND_INPUT_LISTS]
+    # Since tasks can be assigned to any node and we're testing with files, we need to make sure the content is the same
+    # across all nodes.
+    FIRST_INPUT_LIST = ["foo", "bar", "baz"]
+    FIRST_INPUTS = "\n".join(FIRST_INPUT_LIST) + "\n"
+    SECOND_INPUT_LIST = ["razz", "ma", "tazz"]
+    SECOND_INPUTS = "\n".join(SECOND_INPUT_LIST) + "\n"
 
     SCHEMA = { "type": "string", "optional": False }
 
@@ -43,13 +45,7 @@ class CopycatDistributedFileTest(KafkaTest):
             'test' : { 'partitions': 1, 'replication-factor': 1 }
         })
 
-        # FIXME these should have multiple nodes. However, currently the connectors are submitted via command line,
-        # which means we would get duplicates. Both would run, but they would have conflicting keys for offsets and
-        # configs. Until we have real distributed coordination of workers with unified connector submission, we need
-        # to restrict each of these to a single node.
-        self.num_nodes = 1
-        self.source = CopycatDistributedService(test_context, self.num_nodes, self.kafka, [self.INPUT_FILE])
-        self.sink = CopycatDistributedService(test_context, self.num_nodes, self.kafka, [self.OUTPUT_FILE])
+        self.cc = CopycatDistributedService(test_context, 2, self.kafka, [self.INPUT_FILE, self.OUTPUT_FILE])
 
     def test_file_source_and_sink(self, converter="org.apache.kafka.copycat.json.JsonConverter", schemas=True):
         assert converter != None, "converter type must be set"
@@ -58,33 +54,40 @@ class CopycatDistributedFileTest(KafkaTest):
         self.value_converter = converter
         self.schemas = schemas
 
-        # These need to be set
-        self.source.set_configs(self.render("copycat-distributed.properties"), self.render("copycat-file-source.properties"))
-        self.sink.set_configs(self.render("copycat-distributed.properties"), self.render("copycat-file-sink.properties"))
+        self.cc.set_configs(self.render("copycat-distributed.properties"), [self.render("copycat-file-source.properties"), self.render("copycat-file-sink.properties")])
 
-        self.source.start()
-        self.sink.start()
+        self.cc.start()
 
-        # Generating data on the source node should generate new records and create new output on the sink node
-        for node, input in zip(self.source.nodes, self.FIRST_INPUTS):
-            node.account.ssh("echo -e -n " + repr(input) + " >> " + self.INPUT_FILE)
-        wait_until(lambda: self.validate_output(self.FIRST_INPUT_LISTS[:self.num_nodes]), timeout_sec=60, err_msg="Data added to input file was not seen in the output file in a reasonable amount of time.")
+        # Generating data on the source node should generate new records and create new output on the sink node. Timeouts
+        # here need to be more generous than they are for standalone mode because a) it takes longer to write configs,
+        # do rebalancing of the group, etc, and b) without explicit leave group support, rebalancing takes awhile
+        for node in self.cc.nodes:
+            node.account.ssh("echo -e -n " + repr(self.FIRST_INPUTS) + " >> " + self.INPUT_FILE)
+        wait_until(lambda: self.validate_output(self.FIRST_INPUT_LIST), timeout_sec=120, err_msg="Data added to input file was not seen in the output file in a reasonable amount of time.")
 
         # Restarting both should result in them picking up where they left off,
         # only processing new data.
-        self.source.restart()
-        self.sink.restart()
+        self.cc.restart()
 
-        for node, input in zip(self.source.nodes, self.SECOND_INPUTS):
-            node.account.ssh("echo -e -n " + repr(input) + " >> " + self.INPUT_FILE)
-        wait_until(lambda: self.validate_output(self.FIRST_INPUT_LISTS[:self.num_nodes] + self.SECOND_INPUT_LISTS[:self.num_nodes]), timeout_sec=60, err_msg="Sink output file never converged to the same state as the input file")
+        for node in self.cc.nodes:
+            node.account.ssh("echo -e -n " + repr(self.SECOND_INPUTS) + " >> " + self.INPUT_FILE)
+        wait_until(lambda: self.validate_output(self.FIRST_INPUT_LIST + self.SECOND_INPUT_LIST), timeout_sec=120, err_msg="Sink output file never converged to the same state as the input file")
 
-    def validate_output(self, inputs):
+    def validate_output(self, input):
+        input_set = set(input)
+        # Output needs to be collected from all nodes because we can't be sure where the tasks will be scheduled.
+        # Between the first and second rounds, we might even end up with half the data on each node.
+        output_set = set(itertools.chain(*[
+            [line.strip() for line in self.file_contents(node, self.OUTPUT_FILE)] for node in self.cc.nodes
+        ]))
+        #print input_set, output_set
+        return input_set == output_set
+
+
+    def file_contents(self, node, file):
         try:
-            input_set = set(itertools.chain(*inputs))
-            output_set = set(itertools.chain(*[
-                [line.strip() for line in node.account.ssh_capture("cat " + self.OUTPUT_FILE)] for node in self.sink.nodes
-            ]))
-            return input_set == output_set
+            # Convert to a list here or the CalledProcessError may be returned during a call to the generator instead of
+            # immediately
+            return list(node.account.ssh_capture("cat " + file))
         except subprocess.CalledProcessError:
-            return False
+            return []

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/tests/kafkatest/tests/copycat_test.py
----------------------------------------------------------------------
diff --git a/tests/kafkatest/tests/copycat_test.py b/tests/kafkatest/tests/copycat_test.py
index 1bd8ccb..bad5330 100644
--- a/tests/kafkatest/tests/copycat_test.py
+++ b/tests/kafkatest/tests/copycat_test.py
@@ -60,9 +60,8 @@ class CopycatStandaloneFileTest(KafkaTest):
         self.value_converter = converter
         self.schemas = schemas
 
-        # These need to be set
-        self.source.set_configs(self.render("copycat-standalone.properties"), self.render("copycat-file-source.properties"))
-        self.sink.set_configs(self.render("copycat-standalone.properties"), self.render("copycat-file-sink.properties"))
+        self.source.set_configs(self.render("copycat-standalone.properties"), [self.render("copycat-file-source.properties")])
+        self.sink.set_configs(self.render("copycat-standalone.properties"), [self.render("copycat-file-sink.properties")])
 
         self.source.start()
         self.sink.start()

http://git-wip-us.apache.org/repos/asf/kafka/blob/2e617735/tests/kafkatest/tests/templates/copycat-distributed.properties
----------------------------------------------------------------------
diff --git a/tests/kafkatest/tests/templates/copycat-distributed.properties b/tests/kafkatest/tests/templates/copycat-distributed.properties
index 31f9901..325dc85 100644
--- a/tests/kafkatest/tests/templates/copycat-distributed.properties
+++ b/tests/kafkatest/tests/templates/copycat-distributed.properties
@@ -15,6 +15,8 @@
 
 bootstrap.servers={{ kafka.bootstrap_servers() }}
 
+group.id={{ group|default("copycat-cluster") }}
+
 key.converter={{ key_converter|default("org.apache.kafka.copycat.json.JsonConverter") }}
 value.converter={{ value_converter|default("org.apache.kafka.copycat.json.JsonConverter") }}
 {% if key_converter is not defined or key_converter.endswith("JsonConverter") %}
@@ -30,4 +32,7 @@ internal.key.converter.schemas.enable=false
 internal.value.converter.schemas.enable=false
 
 offset.storage.topic={{ OFFSETS_TOPIC }}
-config.storage.topic={{ CONFIG_TOPIC }}
\ No newline at end of file
+config.storage.topic={{ CONFIG_TOPIC }}
+
+# Make sure data gets flushed frequently so tests don't have to wait to ensure they see data in output systems
+offset.flush.interval.ms=5000


Mime
View raw message