kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gwens...@apache.org
Subject [04/26] kafka git commit: KAFKA-2774: Rename Copycat to Kafka Connect
Date Mon, 09 Nov 2015 06:11:43 GMT
http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSinkTaskThreadedTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSinkTaskThreadedTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSinkTaskThreadedTest.java
deleted file mode 100644
index ded78a1..0000000
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSinkTaskThreadedTest.java
+++ /dev/null
@@ -1,563 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * 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;
-
-import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.clients.consumer.OffsetCommitCallback;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.copycat.data.Schema;
-import org.apache.kafka.copycat.data.SchemaAndValue;
-import org.apache.kafka.copycat.errors.CopycatException;
-import org.apache.kafka.copycat.runtime.standalone.StandaloneConfig;
-import org.apache.kafka.copycat.sink.SinkConnector;
-import org.apache.kafka.copycat.sink.SinkRecord;
-import org.apache.kafka.copycat.sink.SinkTask;
-import org.apache.kafka.copycat.storage.Converter;
-import org.apache.kafka.copycat.util.ConnectorTaskId;
-import org.apache.kafka.copycat.util.MockTime;
-import org.apache.kafka.copycat.util.ThreadedTest;
-import org.easymock.Capture;
-import org.easymock.CaptureType;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
-import org.easymock.IExpectationSetters;
-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 org.powermock.reflect.Whitebox;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(WorkerSinkTask.class)
-@PowerMockIgnore("javax.management.*")
-public class WorkerSinkTaskThreadedTest extends ThreadedTest {
-
-    // These are fixed to keep this code simpler. In this example we assume byte[] raw values
-    // with mix of integer/string in Copycat
-    private static final String TOPIC = "test";
-    private static final int PARTITION = 12;
-    private static final int PARTITION2 = 13;
-    private static final int PARTITION3 = 14;
-    private static final long FIRST_OFFSET = 45;
-    private static final Schema KEY_SCHEMA = Schema.INT32_SCHEMA;
-    private static final int KEY = 12;
-    private static final Schema VALUE_SCHEMA = Schema.STRING_SCHEMA;
-    private static final String VALUE = "VALUE";
-    private static final byte[] RAW_KEY = "key".getBytes();
-    private static final byte[] RAW_VALUE = "value".getBytes();
-
-    private static final TopicPartition TOPIC_PARTITION = new TopicPartition(TOPIC, PARTITION);
-    private static final TopicPartition TOPIC_PARTITION2 = new TopicPartition(TOPIC, PARTITION2);
-    private static final TopicPartition TOPIC_PARTITION3 = new TopicPartition(TOPIC, PARTITION3);
-    private static final TopicPartition UNASSIGNED_TOPIC_PARTITION = new TopicPartition(TOPIC, 200);
-
-    private static final Map<String, String> TASK_PROPS = new HashMap<>();
-    static {
-        TASK_PROPS.put(SinkConnector.TOPICS_CONFIG, TOPIC);
-    }
-
-    private ConnectorTaskId taskId = new ConnectorTaskId("job", 0);
-    private Time time;
-    @Mock private SinkTask sinkTask;
-    private Capture<WorkerSinkTaskContext> sinkTaskContext = EasyMock.newCapture();
-    private WorkerConfig workerConfig;
-    @Mock private Converter keyConverter;
-    @Mock
-    private Converter valueConverter;
-    private WorkerSinkTask workerTask;
-    @Mock private KafkaConsumer<byte[], byte[]> consumer;
-    private WorkerSinkTaskThread workerThread;
-    private Capture<ConsumerRebalanceListener> rebalanceListener = EasyMock.newCapture();
-
-    private long recordsReturned;
-
-    @SuppressWarnings("unchecked")
-    @Override
-    public void setup() {
-        super.setup();
-        time = new MockTime();
-        Map<String, String> workerProps = new HashMap<>();
-        workerProps.put("key.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("value.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.key.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.value.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.key.converter.schemas.enable", "false");
-        workerProps.put("internal.value.converter.schemas.enable", "false");
-        workerConfig = new StandaloneConfig(workerProps);
-        workerTask = PowerMock.createPartialMock(
-                WorkerSinkTask.class, new String[]{"createConsumer", "createWorkerThread"},
-                taskId, sinkTask, workerConfig, keyConverter, valueConverter, time);
-
-        recordsReturned = 0;
-    }
-
-    @Test
-    public void testPollsInBackground() throws Exception {
-        expectInitializeTask();
-        Capture<Collection<SinkRecord>> capturedRecords = expectPolls(1L);
-        expectStopTask(10L);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        for (int i = 0; i < 10; i++) {
-            workerThread.iteration();
-        }
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        // Verify contents match expected values, i.e. that they were translated properly. With max
-        // batch size 1 and poll returns 1 message at a time, we should have a matching # of batches
-        assertEquals(10, capturedRecords.getValues().size());
-        int offset = 0;
-        for (Collection<SinkRecord> recs : capturedRecords.getValues()) {
-            assertEquals(1, recs.size());
-            for (SinkRecord rec : recs) {
-                SinkRecord referenceSinkRecord
-                        = new SinkRecord(TOPIC, PARTITION, KEY_SCHEMA, KEY, VALUE_SCHEMA, VALUE, FIRST_OFFSET + offset);
-                assertEquals(referenceSinkRecord, rec);
-                offset++;
-            }
-        }
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommit() throws Exception {
-        expectInitializeTask();
-        // Make each poll() take the offset commit interval
-        Capture<Collection<SinkRecord>> capturedRecords
-                = expectPolls(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT);
-        expectOffsetFlush(1L, null, null, 0, true);
-        expectStopTask(2);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        // First iteration gets one record
-        workerThread.iteration();
-        // Second triggers commit, gets a second offset
-        workerThread.iteration();
-        // Commit finishes synchronously for testing so we can check this immediately
-        assertEquals(0, workerThread.commitFailures());
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        assertEquals(2, capturedRecords.getValues().size());
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommitTaskFlushFailure() throws Exception {
-        expectInitializeTask();
-        Capture<Collection<SinkRecord>> capturedRecords = expectPolls(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT);
-        expectOffsetFlush(1L, new RuntimeException(), null, 0, true);
-        // Should rewind to last known good positions, which in this case will be the offsets loaded during initialization
-        // for all topic partitions
-        consumer.seek(TOPIC_PARTITION, FIRST_OFFSET);
-        PowerMock.expectLastCall();
-        consumer.seek(TOPIC_PARTITION2, FIRST_OFFSET);
-        PowerMock.expectLastCall();
-        consumer.seek(TOPIC_PARTITION3, FIRST_OFFSET);
-        PowerMock.expectLastCall();
-        expectStopTask(2);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        // Second iteration triggers commit
-        workerThread.iteration();
-        workerThread.iteration();
-        assertEquals(1, workerThread.commitFailures());
-        assertEquals(false, Whitebox.getInternalState(workerThread, "committing"));
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommitTaskSuccessAndFlushFailure() throws Exception {
-        // Validate that we rewind to the correct offsets if a task's flush method throws an exception
-
-        expectInitializeTask();
-        Capture<Collection<SinkRecord>> capturedRecords = expectPolls(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT);
-        expectOffsetFlush(1L, null, null, 0, true);
-        expectOffsetFlush(2L, new RuntimeException(), null, 0, true);
-        // Should rewind to last known committed positions
-        consumer.seek(TOPIC_PARTITION, FIRST_OFFSET + 1);
-        PowerMock.expectLastCall();
-        consumer.seek(TOPIC_PARTITION2, FIRST_OFFSET);
-        PowerMock.expectLastCall();
-        consumer.seek(TOPIC_PARTITION3, FIRST_OFFSET);
-        PowerMock.expectLastCall();
-        expectStopTask(2);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        // Second iteration triggers first commit, third iteration triggers second (failing) commit
-        workerThread.iteration();
-        workerThread.iteration();
-        workerThread.iteration();
-        assertEquals(1, workerThread.commitFailures());
-        assertEquals(false, Whitebox.getInternalState(workerThread, "committing"));
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommitConsumerFailure() throws Exception {
-        expectInitializeTask();
-        Capture<Collection<SinkRecord>> capturedRecords
-                = expectPolls(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT);
-        expectOffsetFlush(1L, null, new Exception(), 0, true);
-        expectStopTask(2);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        // Second iteration triggers commit
-        workerThread.iteration();
-        workerThread.iteration();
-        // TODO Response to consistent failures?
-        assertEquals(1, workerThread.commitFailures());
-        assertEquals(false, Whitebox.getInternalState(workerThread, "committing"));
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommitTimeout() throws Exception {
-        expectInitializeTask();
-        // Cut down amount of time to pass in each poll so we trigger exactly 1 offset commit
-        Capture<Collection<SinkRecord>> capturedRecords
-                = expectPolls(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_DEFAULT / 2);
-        expectOffsetFlush(2L, null, null, WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_DEFAULT, false);
-        expectStopTask(4);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        // Third iteration triggers commit, fourth gives a chance to trigger the timeout but doesn't
-        // trigger another commit
-        workerThread.iteration();
-        workerThread.iteration();
-        workerThread.iteration();
-        workerThread.iteration();
-        // TODO Response to consistent failures?
-        assertEquals(1, workerThread.commitFailures());
-        assertEquals(false, Whitebox.getInternalState(workerThread, "committing"));
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testAssignmentPauseResume() throws Exception {
-        // Just validate that the calls are passed through to the consumer, and that where appropriate errors are
-        // converted
-        expectInitializeTask();
-
-        expectOnePoll().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                assertEquals(new HashSet<>(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3)),
-                        sinkTaskContext.getValue().assignment());
-                return null;
-            }
-        });
-        EasyMock.expect(consumer.assignment()).andReturn(new HashSet<>(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3)));
-
-        expectOnePoll().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                try {
-                    sinkTaskContext.getValue().pause(UNASSIGNED_TOPIC_PARTITION);
-                    fail("Trying to pause unassigned partition should have thrown an Copycat exception");
-                } catch (CopycatException e) {
-                    // expected
-                }
-                sinkTaskContext.getValue().pause(TOPIC_PARTITION, TOPIC_PARTITION2);
-                return null;
-            }
-        });
-        consumer.pause(UNASSIGNED_TOPIC_PARTITION);
-        PowerMock.expectLastCall().andThrow(new IllegalStateException("unassigned topic partition"));
-        consumer.pause(TOPIC_PARTITION, TOPIC_PARTITION2);
-        PowerMock.expectLastCall();
-
-        expectOnePoll().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                try {
-                    sinkTaskContext.getValue().resume(UNASSIGNED_TOPIC_PARTITION);
-                    fail("Trying to resume unassigned partition should have thrown an Copycat exception");
-                } catch (CopycatException e) {
-                    // expected
-                }
-
-                sinkTaskContext.getValue().resume(TOPIC_PARTITION, TOPIC_PARTITION2);
-                return null;
-            }
-        });
-        consumer.resume(UNASSIGNED_TOPIC_PARTITION);
-        PowerMock.expectLastCall().andThrow(new IllegalStateException("unassigned topic partition"));
-        consumer.resume(TOPIC_PARTITION, TOPIC_PARTITION2);
-        PowerMock.expectLastCall();
-
-        expectStopTask(0);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        workerThread.iteration();
-        workerThread.iteration();
-        workerThread.iteration();
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testRewind() throws Exception {
-        expectInitializeTask();
-        final long startOffset = 40L;
-        final Map<TopicPartition, Long> offsets = new HashMap<>();
-
-        expectOnePoll().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                offsets.put(TOPIC_PARTITION, startOffset);
-                sinkTaskContext.getValue().offset(offsets);
-                return null;
-            }
-        });
-
-        consumer.seek(TOPIC_PARTITION, startOffset);
-        EasyMock.expectLastCall();
-
-        expectOnePoll().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                Map<TopicPartition, Long> offsets = sinkTaskContext.getValue().offsets();
-                assertEquals(0, offsets.size());
-                return null;
-            }
-        });
-
-        expectStopTask(3);
-        EasyMock.expect(workerThread.awaitShutdown(EasyMock.anyLong(), EasyMock.<TimeUnit>anyObject())).andReturn(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(TASK_PROPS);
-        workerTask.joinConsumerGroupAndStart();
-        workerThread.iteration();
-        workerThread.iteration();
-        workerTask.stop();
-        workerTask.awaitStop(Long.MAX_VALUE);
-        workerTask.close();
-
-        PowerMock.verifyAll();
-    }
-
-    private void expectInitializeTask() throws Exception {
-        PowerMock.expectPrivate(workerTask, "createConsumer").andReturn(consumer);
-
-        workerThread = PowerMock.createPartialMock(WorkerSinkTaskThread.class, new String[]{"start", "awaitShutdown"},
-                workerTask, "mock-worker-thread", time,
-                workerConfig);
-        PowerMock.expectPrivate(workerTask, "createWorkerThread")
-                .andReturn(workerThread);
-        workerThread.start();
-        PowerMock.expectLastCall();
-
-        consumer.subscribe(EasyMock.eq(Arrays.asList(TOPIC)), EasyMock.capture(rebalanceListener));
-        PowerMock.expectLastCall();
-
-        EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(new IAnswer<ConsumerRecords<byte[], byte[]>>() {
-            @Override
-            public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
-                rebalanceListener.getValue().onPartitionsAssigned(Arrays.asList(TOPIC_PARTITION, TOPIC_PARTITION2, TOPIC_PARTITION3));
-                return ConsumerRecords.empty();
-            }
-        });
-        EasyMock.expect(consumer.position(TOPIC_PARTITION)).andReturn(FIRST_OFFSET);
-        EasyMock.expect(consumer.position(TOPIC_PARTITION2)).andReturn(FIRST_OFFSET);
-        EasyMock.expect(consumer.position(TOPIC_PARTITION3)).andReturn(FIRST_OFFSET);
-
-        sinkTask.initialize(EasyMock.capture(sinkTaskContext));
-        PowerMock.expectLastCall();
-        sinkTask.start(TASK_PROPS);
-        PowerMock.expectLastCall();
-    }
-
-    private void expectStopTask(final long expectedMessages) throws Exception {
-        final long finalOffset = FIRST_OFFSET + expectedMessages - 1;
-
-        sinkTask.stop();
-        PowerMock.expectLastCall();
-
-        // No offset commit since it happens in the mocked worker thread, but the main thread does need to wake up the
-        // consumer so it exits quickly
-        consumer.wakeup();
-        PowerMock.expectLastCall();
-
-        consumer.close();
-        PowerMock.expectLastCall();
-    }
-
-    // Note that this can only be called once per test currently
-    private Capture<Collection<SinkRecord>> expectPolls(final long pollDelayMs) throws Exception {
-        // Stub out all the consumer stream/iterator responses, which we just want to verify occur,
-        // but don't care about the exact details here.
-        EasyMock.expect(consumer.poll(EasyMock.anyLong())).andStubAnswer(
-                new IAnswer<ConsumerRecords<byte[], byte[]>>() {
-                    @Override
-                    public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
-                        // "Sleep" so time will progress
-                        time.sleep(pollDelayMs);
-                        ConsumerRecords<byte[], byte[]> records = new ConsumerRecords<>(
-                                Collections.singletonMap(
-                                        new TopicPartition(TOPIC, PARTITION),
-                                        Arrays.asList(
-                                                new ConsumerRecord<>(TOPIC, PARTITION, FIRST_OFFSET + recordsReturned, RAW_KEY, RAW_VALUE)
-                                        )));
-                        recordsReturned++;
-                        return records;
-                    }
-                });
-        EasyMock.expect(keyConverter.toCopycatData(TOPIC, RAW_KEY)).andReturn(new SchemaAndValue(KEY_SCHEMA, KEY)).anyTimes();
-        EasyMock.expect(valueConverter.toCopycatData(TOPIC, RAW_VALUE)).andReturn(new SchemaAndValue(VALUE_SCHEMA, VALUE)).anyTimes();
-        Capture<Collection<SinkRecord>> capturedRecords = EasyMock.newCapture(CaptureType.ALL);
-        sinkTask.put(EasyMock.capture(capturedRecords));
-        EasyMock.expectLastCall().anyTimes();
-        return capturedRecords;
-    }
-
-    private IExpectationSetters<Object> expectOnePoll() {
-        // Currently the SinkTask's put() method will not be invoked unless we provide some data, so instead of
-        // returning empty data, we return one record. The expectation is that the data will be ignored by the
-        // response behavior specified using the return value of this method.
-        EasyMock.expect(consumer.poll(EasyMock.anyLong())).andAnswer(
-                new IAnswer<ConsumerRecords<byte[], byte[]>>() {
-                    @Override
-                    public ConsumerRecords<byte[], byte[]> answer() throws Throwable {
-                        // "Sleep" so time will progress
-                        time.sleep(1L);
-                        ConsumerRecords<byte[], byte[]> records = new ConsumerRecords<>(
-                                Collections.singletonMap(
-                                        new TopicPartition(TOPIC, PARTITION),
-                                        Arrays.asList(
-                                                new ConsumerRecord<>(TOPIC, PARTITION, FIRST_OFFSET + recordsReturned, RAW_KEY, RAW_VALUE)
-                                        )));
-                        recordsReturned++;
-                        return records;
-                    }
-                });
-        EasyMock.expect(keyConverter.toCopycatData(TOPIC, RAW_KEY)).andReturn(new SchemaAndValue(KEY_SCHEMA, KEY));
-        EasyMock.expect(valueConverter.toCopycatData(TOPIC, RAW_VALUE)).andReturn(new SchemaAndValue(VALUE_SCHEMA, VALUE));
-        sinkTask.put(EasyMock.anyObject(Collection.class));
-        return EasyMock.expectLastCall();
-    }
-
-    private Capture<OffsetCommitCallback> expectOffsetFlush(final long expectedMessages,
-                                                              final RuntimeException flushError,
-                                                              final Exception consumerCommitError,
-                                                              final long consumerCommitDelayMs,
-                                                              final boolean invokeCallback)
-            throws Exception {
-        final long finalOffset = FIRST_OFFSET + expectedMessages;
-
-        // All assigned partitions will have offsets committed, but we've only processed messages/updated offsets for one
-        final Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
-        offsetsToCommit.put(TOPIC_PARTITION, new OffsetAndMetadata(finalOffset));
-        offsetsToCommit.put(TOPIC_PARTITION2, new OffsetAndMetadata(FIRST_OFFSET));
-        offsetsToCommit.put(TOPIC_PARTITION3, new OffsetAndMetadata(FIRST_OFFSET));
-        sinkTask.flush(offsetsToCommit);
-        IExpectationSetters<Object> flushExpectation = PowerMock.expectLastCall();
-        if (flushError != null) {
-            flushExpectation.andThrow(flushError).once();
-            return null;
-        }
-
-        final Capture<OffsetCommitCallback> capturedCallback = EasyMock.newCapture();
-        consumer.commitAsync(EasyMock.eq(offsetsToCommit),
-                EasyMock.capture(capturedCallback));
-        PowerMock.expectLastCall().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                time.sleep(consumerCommitDelayMs);
-                if (invokeCallback)
-                    capturedCallback.getValue().onComplete(offsetsToCommit, consumerCommitError);
-                return null;
-            }
-        });
-        return capturedCallback;
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSourceTaskTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSourceTaskTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSourceTaskTest.java
deleted file mode 100644
index 0fa14bd..0000000
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerSourceTaskTest.java
+++ /dev/null
@@ -1,308 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * 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;
-
-import org.apache.kafka.common.utils.SystemTime;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.utils.Utils;
-import org.apache.kafka.copycat.data.Schema;
-import org.apache.kafka.copycat.runtime.standalone.StandaloneConfig;
-import org.apache.kafka.copycat.source.SourceRecord;
-import org.apache.kafka.copycat.source.SourceTask;
-import org.apache.kafka.copycat.source.SourceTaskContext;
-import org.apache.kafka.copycat.storage.Converter;
-import org.apache.kafka.copycat.storage.OffsetStorageReader;
-import org.apache.kafka.copycat.storage.OffsetStorageWriter;
-import org.apache.kafka.copycat.util.Callback;
-import org.apache.kafka.copycat.util.ConnectorTaskId;
-import org.apache.kafka.copycat.util.ThreadedTest;
-import org.easymock.Capture;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
-import org.easymock.IExpectationSetters;
-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.modules.junit4.PowerMockRunner;
-import org.powermock.reflect.Whitebox;
-
-import java.util.*;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-
-import static org.junit.Assert.*;
-
-@RunWith(PowerMockRunner.class)
-public class WorkerSourceTaskTest extends ThreadedTest {
-    private static final String TOPIC = "topic";
-    private static final Map<String, byte[]> PARTITION = Collections.singletonMap("key", "partition".getBytes());
-    private static final Map<String, Integer> OFFSET = Collections.singletonMap("key", 12);
-
-    // Copycat-format data
-    private static final Schema KEY_SCHEMA = Schema.INT32_SCHEMA;
-    private static final Integer KEY = -1;
-    private static final Schema RECORD_SCHEMA = Schema.INT64_SCHEMA;
-    private static final Long RECORD = 12L;
-    // Serialized data. The actual format of this data doesn't matter -- we just want to see that the right version
-    // is used in the right place.
-    private static final byte[] SERIALIZED_KEY = "converted-key".getBytes();
-    private static final byte[] SERIALIZED_RECORD = "converted-record".getBytes();
-
-    private ConnectorTaskId taskId = new ConnectorTaskId("job", 0);
-    private WorkerConfig config;
-    @Mock private SourceTask sourceTask;
-    @Mock private Converter keyConverter;
-    @Mock private Converter valueConverter;
-    @Mock private KafkaProducer<byte[], byte[]> producer;
-    @Mock private OffsetStorageReader offsetReader;
-    @Mock private OffsetStorageWriter offsetWriter;
-    private WorkerSourceTask workerTask;
-    @Mock private Future<RecordMetadata> sendFuture;
-
-    private Capture<org.apache.kafka.clients.producer.Callback> producerCallbacks;
-
-    private static final Map<String, String> EMPTY_TASK_PROPS = Collections.emptyMap();
-    private static final List<SourceRecord> RECORDS = Arrays.asList(
-            new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD)
-    );
-
-    @Override
-    public void setup() {
-        super.setup();
-        Map<String, String> workerProps = new HashMap<>();
-        workerProps.put("key.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("value.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.key.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.value.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.key.converter.schemas.enable", "false");
-        workerProps.put("internal.value.converter.schemas.enable", "false");
-        config = new StandaloneConfig(workerProps);
-        producerCallbacks = EasyMock.newCapture();
-    }
-
-    private void createWorkerTask() {
-        workerTask = new WorkerSourceTask(taskId, sourceTask, keyConverter, valueConverter, producer,
-                offsetReader, offsetWriter, config, new SystemTime());
-    }
-
-    @Test
-    public void testPollsInBackground() throws Exception {
-        createWorkerTask();
-
-        sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
-        EasyMock.expectLastCall();
-        sourceTask.start(EMPTY_TASK_PROPS);
-        EasyMock.expectLastCall();
-
-        final CountDownLatch pollLatch = expectPolls(10);
-        // In this test, we don't flush, so nothing goes any further than the offset writer
-
-        sourceTask.stop();
-        EasyMock.expectLastCall();
-        expectOffsetFlush(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(EMPTY_TASK_PROPS);
-        awaitPolls(pollLatch);
-        workerTask.stop();
-        assertEquals(true, workerTask.awaitStop(1000));
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommit() throws Exception {
-        // Test that the task commits properly when prompted
-        createWorkerTask();
-
-        sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
-        EasyMock.expectLastCall();
-        sourceTask.start(EMPTY_TASK_PROPS);
-        EasyMock.expectLastCall();
-
-        // We'll wait for some data, then trigger a flush
-        final CountDownLatch pollLatch = expectPolls(1);
-        expectOffsetFlush(true);
-
-        sourceTask.stop();
-        EasyMock.expectLastCall();
-        expectOffsetFlush(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(EMPTY_TASK_PROPS);
-        awaitPolls(pollLatch);
-        assertTrue(workerTask.commitOffsets());
-        workerTask.stop();
-        assertEquals(true, workerTask.awaitStop(1000));
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCommitFailure() throws Exception {
-        // Test that the task commits properly when prompted
-        createWorkerTask();
-
-        sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
-        EasyMock.expectLastCall();
-        sourceTask.start(EMPTY_TASK_PROPS);
-        EasyMock.expectLastCall();
-
-        // We'll wait for some data, then trigger a flush
-        final CountDownLatch pollLatch = expectPolls(1);
-        expectOffsetFlush(false);
-
-        sourceTask.stop();
-        EasyMock.expectLastCall();
-        expectOffsetFlush(true);
-
-        PowerMock.replayAll();
-
-        workerTask.start(EMPTY_TASK_PROPS);
-        awaitPolls(pollLatch);
-        assertFalse(workerTask.commitOffsets());
-        workerTask.stop();
-        assertEquals(true, workerTask.awaitStop(1000));
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testSendRecordsConvertsData() throws Exception {
-        createWorkerTask();
-
-        List<SourceRecord> records = new ArrayList<>();
-        // Can just use the same record for key and value
-        records.add(new SourceRecord(PARTITION, OFFSET, "topic", null, KEY_SCHEMA, KEY, RECORD_SCHEMA, RECORD));
-
-        Capture<ProducerRecord<byte[], byte[]>> sent = expectSendRecord();
-
-        PowerMock.replayAll();
-
-        Whitebox.invokeMethod(workerTask, "sendRecords", records);
-        assertEquals(SERIALIZED_KEY, sent.getValue().key());
-        assertEquals(SERIALIZED_RECORD, sent.getValue().value());
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testSlowTaskStart() throws Exception {
-        createWorkerTask();
-
-        sourceTask.initialize(EasyMock.anyObject(SourceTaskContext.class));
-        EasyMock.expectLastCall();
-        sourceTask.start(EMPTY_TASK_PROPS);
-        EasyMock.expectLastCall().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                Utils.sleep(100);
-                return null;
-            }
-        });
-        sourceTask.stop();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        workerTask.start(EMPTY_TASK_PROPS);
-        // Stopping immediately while the other thread has work to do should result in no polling, no offset commits,
-        // exiting the work thread immediately, and the stop() method will be invoked in the background thread since it
-        // cannot be invoked immediately in the thread trying to stop the task.
-        workerTask.stop();
-        assertEquals(true, workerTask.awaitStop(1000));
-
-        PowerMock.verifyAll();
-    }
-
-    private CountDownLatch expectPolls(int count) throws InterruptedException {
-        final CountDownLatch latch = new CountDownLatch(count);
-        // Note that we stub these to allow any number of calls because the thread will continue to
-        // run. The count passed in + latch returned just makes sure we get *at least* that number of
-        // calls
-        EasyMock.expect(sourceTask.poll())
-                .andStubAnswer(new IAnswer<List<SourceRecord>>() {
-                    @Override
-                    public List<SourceRecord> answer() throws Throwable {
-                        latch.countDown();
-                        return RECORDS;
-                    }
-                });
-        // Fallout of the poll() call
-        expectSendRecord();
-        return latch;
-    }
-
-    private Capture<ProducerRecord<byte[], byte[]>> expectSendRecord() throws InterruptedException {
-        EasyMock.expect(keyConverter.fromCopycatData(TOPIC, KEY_SCHEMA, KEY)).andStubReturn(SERIALIZED_KEY);
-        EasyMock.expect(valueConverter.fromCopycatData(TOPIC, RECORD_SCHEMA, RECORD)).andStubReturn(SERIALIZED_RECORD);
-
-        Capture<ProducerRecord<byte[], byte[]>> sent = EasyMock.newCapture();
-        // 1. Converted data passed to the producer, which will need callbacks invoked for flush to work
-        EasyMock.expect(
-                producer.send(EasyMock.capture(sent),
-                        EasyMock.capture(producerCallbacks)))
-                .andStubAnswer(new IAnswer<Future<RecordMetadata>>() {
-                    @Override
-                    public Future<RecordMetadata> answer() throws Throwable {
-                        synchronized (producerCallbacks) {
-                            for (org.apache.kafka.clients.producer.Callback cb : producerCallbacks.getValues()) {
-                                cb.onCompletion(new RecordMetadata(new TopicPartition("foo", 0), 0, 0), null);
-                            }
-                            producerCallbacks.reset();
-                        }
-                        return sendFuture;
-                    }
-                });
-        // 2. Offset data is passed to the offset storage.
-        offsetWriter.offset(PARTITION, OFFSET);
-        PowerMock.expectLastCall().anyTimes();
-
-        return sent;
-    }
-
-    private void awaitPolls(CountDownLatch latch) throws InterruptedException {
-        latch.await(1000, TimeUnit.MILLISECONDS);
-    }
-
-    @SuppressWarnings("unchecked")
-    private void expectOffsetFlush(boolean succeed) throws Exception {
-        EasyMock.expect(offsetWriter.beginFlush()).andReturn(true);
-        Future<Void> flushFuture = PowerMock.createMock(Future.class);
-        EasyMock.expect(offsetWriter.doFlush(EasyMock.anyObject(Callback.class))).andReturn(flushFuture);
-        // Should throw for failure
-        IExpectationSetters<Void> futureGetExpect = EasyMock.expect(
-                flushFuture.get(EasyMock.anyLong(), EasyMock.anyObject(TimeUnit.class)));
-        if (succeed) {
-            futureGetExpect.andReturn(null);
-        } else {
-            futureGetExpect.andThrow(new TimeoutException());
-            offsetWriter.cancelFlush();
-            PowerMock.expectLastCall();
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerTest.java
----------------------------------------------------------------------
diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerTest.java
deleted file mode 100644
index 00cef2b..0000000
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/WorkerTest.java
+++ /dev/null
@@ -1,397 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- * <p/>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p/>
- * 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;
-
-import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.copycat.connector.Connector;
-import org.apache.kafka.copycat.connector.ConnectorContext;
-import org.apache.kafka.copycat.connector.Task;
-import org.apache.kafka.copycat.errors.CopycatException;
-import org.apache.kafka.copycat.runtime.standalone.StandaloneConfig;
-import org.apache.kafka.copycat.sink.SinkTask;
-import org.apache.kafka.copycat.source.SourceRecord;
-import org.apache.kafka.copycat.source.SourceTask;
-import org.apache.kafka.copycat.storage.Converter;
-import org.apache.kafka.copycat.storage.OffsetBackingStore;
-import org.apache.kafka.copycat.storage.OffsetStorageReader;
-import org.apache.kafka.copycat.storage.OffsetStorageWriter;
-import org.apache.kafka.copycat.util.ConnectorTaskId;
-import org.apache.kafka.copycat.util.MockTime;
-import org.apache.kafka.copycat.util.ThreadedTest;
-import org.easymock.EasyMock;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.powermock.api.easymock.PowerMock;
-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.HashSet;
-import java.util.List;
-import java.util.Map;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(Worker.class)
-@PowerMockIgnore("javax.management.*")
-public class WorkerTest extends ThreadedTest {
-
-    private static final String CONNECTOR_ID = "test-connector";
-    private static final ConnectorTaskId TASK_ID = new ConnectorTaskId("job", 0);
-
-    private WorkerConfig config;
-    private Worker worker;
-    private OffsetBackingStore offsetBackingStore = PowerMock.createMock(OffsetBackingStore.class);
-
-    @Before
-    public void setup() {
-        super.setup();
-
-        Map<String, String> workerProps = new HashMap<>();
-        workerProps.put("key.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("value.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.key.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.value.converter", "org.apache.kafka.copycat.json.JsonConverter");
-        workerProps.put("internal.key.converter.schemas.enable", "false");
-        workerProps.put("internal.value.converter.schemas.enable", "false");
-        config = new StandaloneConfig(workerProps);
-    }
-
-    @Test
-    public void testAddRemoveConnector() throws Exception {
-        offsetBackingStore.configure(EasyMock.anyObject(Map.class));
-        EasyMock.expectLastCall();
-        offsetBackingStore.start();
-        EasyMock.expectLastCall();
-
-        // Create
-        Connector connector = PowerMock.createMock(Connector.class);
-        ConnectorContext ctx = PowerMock.createMock(ConnectorContext.class);
-
-        PowerMock.mockStatic(Worker.class);
-        PowerMock.expectPrivate(Worker.class, "instantiateConnector", new Object[]{TestConnector.class}).andReturn(connector);
-        EasyMock.expect(connector.version()).andReturn("1.0");
-
-        Map<String, String> props = new HashMap<>();
-        props.put(ConnectorConfig.TOPICS_CONFIG, "foo,bar");
-        props.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
-        props.put(ConnectorConfig.NAME_CONFIG, CONNECTOR_ID);
-        props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestConnector.class.getName());
-
-        connector.initialize(ctx);
-        EasyMock.expectLastCall();
-        connector.start(props);
-        EasyMock.expectLastCall();
-
-        // Remove
-        connector.stop();
-        EasyMock.expectLastCall();
-
-        offsetBackingStore.stop();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        worker = new Worker(new MockTime(), config, offsetBackingStore);
-        worker.start();
-
-        ConnectorConfig config = new ConnectorConfig(props);
-        assertEquals(Collections.emptySet(), worker.connectorNames());
-        worker.addConnector(config, ctx);
-        assertEquals(new HashSet<>(Arrays.asList(CONNECTOR_ID)), worker.connectorNames());
-        try {
-            worker.addConnector(config, ctx);
-            fail("Should have thrown exception when trying to add connector with same name.");
-        } catch (CopycatException e) {
-            // expected
-        }
-        worker.stopConnector(CONNECTOR_ID);
-        assertEquals(Collections.emptySet(), worker.connectorNames());
-        // Nothing should be left, so this should effectively be a nop
-        worker.stop();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test(expected = CopycatException.class)
-    public void testStopInvalidConnector() {
-        offsetBackingStore.configure(EasyMock.anyObject(Map.class));
-        EasyMock.expectLastCall();
-        offsetBackingStore.start();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        worker = new Worker(new MockTime(), config, offsetBackingStore);
-        worker.start();
-
-        worker.stopConnector(CONNECTOR_ID);
-    }
-
-    @Test
-    public void testReconfigureConnectorTasks() throws Exception {
-        offsetBackingStore.configure(EasyMock.anyObject(Map.class));
-        EasyMock.expectLastCall();
-        offsetBackingStore.start();
-        EasyMock.expectLastCall();
-
-        // Create
-        Connector connector = PowerMock.createMock(Connector.class);
-        ConnectorContext ctx = PowerMock.createMock(ConnectorContext.class);
-
-        PowerMock.mockStatic(Worker.class);
-        PowerMock.expectPrivate(Worker.class, "instantiateConnector", new Object[]{TestConnector.class}).andReturn(connector);
-        EasyMock.expect(connector.version()).andReturn("1.0");
-
-        Map<String, String> props = new HashMap<>();
-        props.put(ConnectorConfig.TOPICS_CONFIG, "foo,bar");
-        props.put(ConnectorConfig.TASKS_MAX_CONFIG, "1");
-        props.put(ConnectorConfig.NAME_CONFIG, CONNECTOR_ID);
-        props.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, TestConnector.class.getName());
-
-        connector.initialize(ctx);
-        EasyMock.expectLastCall();
-        connector.start(props);
-        EasyMock.expectLastCall();
-
-        // Reconfigure
-        EasyMock.<Class<? extends Task>>expect(connector.taskClass()).andReturn(TestSourceTask.class);
-        Map<String, String> taskProps = new HashMap<>();
-        taskProps.put("foo", "bar");
-        EasyMock.expect(connector.taskConfigs(2)).andReturn(Arrays.asList(taskProps, taskProps));
-
-        // Remove
-        connector.stop();
-        EasyMock.expectLastCall();
-
-        offsetBackingStore.stop();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        worker = new Worker(new MockTime(), config, offsetBackingStore);
-        worker.start();
-
-        ConnectorConfig config = new ConnectorConfig(props);
-        assertEquals(Collections.emptySet(), worker.connectorNames());
-        worker.addConnector(config, ctx);
-        assertEquals(new HashSet<>(Arrays.asList(CONNECTOR_ID)), worker.connectorNames());
-        try {
-            worker.addConnector(config, ctx);
-            fail("Should have thrown exception when trying to add connector with same name.");
-        } catch (CopycatException e) {
-            // expected
-        }
-        List<Map<String, String>> taskConfigs = worker.connectorTaskConfigs(CONNECTOR_ID, 2, Arrays.asList("foo", "bar"));
-        Map<String, String> expectedTaskProps = new HashMap<>();
-        expectedTaskProps.put("foo", "bar");
-        expectedTaskProps.put(TaskConfig.TASK_CLASS_CONFIG, TestSourceTask.class.getName());
-        expectedTaskProps.put(SinkTask.TOPICS_CONFIG, "foo,bar");
-        assertEquals(2, taskConfigs.size());
-        assertEquals(expectedTaskProps, taskConfigs.get(0));
-        assertEquals(expectedTaskProps, taskConfigs.get(1));
-        worker.stopConnector(CONNECTOR_ID);
-        assertEquals(Collections.emptySet(), worker.connectorNames());
-        // Nothing should be left, so this should effectively be a nop
-        worker.stop();
-
-        PowerMock.verifyAll();
-    }
-
-
-    @Test
-    public void testAddRemoveTask() throws Exception {
-        offsetBackingStore.configure(EasyMock.anyObject(Map.class));
-        EasyMock.expectLastCall();
-        offsetBackingStore.start();
-        EasyMock.expectLastCall();
-
-        ConnectorTaskId taskId = new ConnectorTaskId("job", 0);
-
-        // Create
-        TestSourceTask task = PowerMock.createMock(TestSourceTask.class);
-        WorkerSourceTask workerTask = PowerMock.createMock(WorkerSourceTask.class);
-
-        PowerMock.mockStatic(Worker.class);
-        PowerMock.expectPrivate(Worker.class, "instantiateTask", new Object[]{TestSourceTask.class}).andReturn(task);
-        EasyMock.expect(task.version()).andReturn("1.0");
-
-        PowerMock.expectNew(
-                WorkerSourceTask.class, EasyMock.eq(taskId), EasyMock.eq(task),
-                EasyMock.anyObject(Converter.class),
-                EasyMock.anyObject(Converter.class),
-                EasyMock.anyObject(KafkaProducer.class),
-                EasyMock.anyObject(OffsetStorageReader.class),
-                EasyMock.anyObject(OffsetStorageWriter.class),
-                EasyMock.anyObject(WorkerConfig.class),
-                EasyMock.anyObject(Time.class))
-                .andReturn(workerTask);
-        Map<String, String> origProps = new HashMap<>();
-        origProps.put(TaskConfig.TASK_CLASS_CONFIG, TestSourceTask.class.getName());
-        workerTask.start(origProps);
-        EasyMock.expectLastCall();
-
-        // Remove
-        workerTask.stop();
-        EasyMock.expectLastCall();
-        EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andStubReturn(true);
-        workerTask.close();
-        EasyMock.expectLastCall();
-
-        offsetBackingStore.stop();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        worker = new Worker(new MockTime(), config, offsetBackingStore);
-        worker.start();
-        assertEquals(Collections.emptySet(), worker.taskIds());
-        worker.addTask(taskId, new TaskConfig(origProps));
-        assertEquals(new HashSet<>(Arrays.asList(taskId)), worker.taskIds());
-        worker.stopTask(taskId);
-        assertEquals(Collections.emptySet(), worker.taskIds());
-        // Nothing should be left, so this should effectively be a nop
-        worker.stop();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test(expected = CopycatException.class)
-    public void testStopInvalidTask() {
-        offsetBackingStore.configure(EasyMock.anyObject(Map.class));
-        EasyMock.expectLastCall();
-        offsetBackingStore.start();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        worker = new Worker(new MockTime(), config, offsetBackingStore);
-        worker.start();
-
-        worker.stopTask(TASK_ID);
-    }
-
-    @Test
-    public void testCleanupTasksOnStop() throws Exception {
-        offsetBackingStore.configure(EasyMock.anyObject(Map.class));
-        EasyMock.expectLastCall();
-        offsetBackingStore.start();
-        EasyMock.expectLastCall();
-
-        // Create
-        TestSourceTask task = PowerMock.createMock(TestSourceTask.class);
-        WorkerSourceTask workerTask = PowerMock.createMock(WorkerSourceTask.class);
-
-        PowerMock.mockStatic(Worker.class);
-        PowerMock.expectPrivate(Worker.class, "instantiateTask", new Object[]{TestSourceTask.class}).andReturn(task);
-        EasyMock.expect(task.version()).andReturn("1.0");
-        
-        PowerMock.expectNew(
-                WorkerSourceTask.class, EasyMock.eq(TASK_ID), EasyMock.eq(task),
-                EasyMock.anyObject(Converter.class),
-                EasyMock.anyObject(Converter.class),
-                EasyMock.anyObject(KafkaProducer.class),
-                EasyMock.anyObject(OffsetStorageReader.class),
-                EasyMock.anyObject(OffsetStorageWriter.class),
-                EasyMock.anyObject(WorkerConfig.class),
-                EasyMock.anyObject(Time.class))
-                .andReturn(workerTask);
-        Map<String, String> origProps = new HashMap<>();
-        origProps.put(TaskConfig.TASK_CLASS_CONFIG, TestSourceTask.class.getName());
-        workerTask.start(origProps);
-        EasyMock.expectLastCall();
-
-        // Remove on Worker.stop()
-        workerTask.stop();
-        EasyMock.expectLastCall();
-        EasyMock.expect(workerTask.awaitStop(EasyMock.anyLong())).andReturn(true);
-        // Note that in this case we *do not* commit offsets since it's an unclean shutdown
-        workerTask.close();
-        EasyMock.expectLastCall();
-
-        offsetBackingStore.stop();
-        EasyMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        worker = new Worker(new MockTime(), config, offsetBackingStore);
-        worker.start();
-        worker.addTask(TASK_ID, new TaskConfig(origProps));
-        worker.stop();
-
-        PowerMock.verifyAll();
-    }
-
-
-    private static class TestConnector extends Connector {
-        @Override
-        public String version() {
-            return "1.0";
-        }
-
-        @Override
-        public void start(Map<String, String> props) {
-
-        }
-
-        @Override
-        public Class<? extends Task> taskClass() {
-            return null;
-        }
-
-        @Override
-        public List<Map<String, String>> taskConfigs(int maxTasks) {
-            return null;
-        }
-
-        @Override
-        public void stop() {
-
-        }
-    }
-
-    private static class TestSourceTask extends SourceTask {
-        public TestSourceTask() {
-        }
-
-        @Override
-        public String version() {
-            return "1.0";
-        }
-
-        @Override
-        public void start(Map<String, String> props) {
-        }
-
-        @Override
-        public List<SourceRecord> poll() throws InterruptedException {
-            return null;
-        }
-
-        @Override
-        public void stop() {
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/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
deleted file mode 100644
index 512cb5c..0000000
--- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/runtime/distributed/DistributedHerderTest.java
+++ /dev/null
@@ -1,573 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- **/
-
-package org.apache.kafka.copycat.runtime.distributed;
-
-import org.apache.kafka.clients.CommonClientConfigs;
-import org.apache.kafka.common.utils.MockTime;
-import org.apache.kafka.copycat.connector.ConnectorContext;
-import org.apache.kafka.copycat.errors.AlreadyExistsException;
-import org.apache.kafka.copycat.runtime.ConnectorConfig;
-import org.apache.kafka.copycat.runtime.Herder;
-import org.apache.kafka.copycat.runtime.TaskConfig;
-import org.apache.kafka.copycat.runtime.Worker;
-import org.apache.kafka.copycat.runtime.WorkerConfig;
-import org.apache.kafka.copycat.runtime.rest.entities.ConnectorInfo;
-import org.apache.kafka.copycat.runtime.rest.entities.TaskInfo;
-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.Capture;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
-import org.junit.Before;
-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 org.powermock.reflect.Whitebox;
-
-import java.util.ArrayList;
-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;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(DistributedHerder.class)
-@PowerMockIgnore("javax.management.*")
-public class DistributedHerderTest {
-    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(DistributedConfig.GROUP_ID_CONFIG, "test-copycat-group");
-        // The WorkerConfig base class has some required settings without defaults
-        HERDER_CONFIG.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.copycat.json.JsonConverter");
-        HERDER_CONFIG.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.copycat.json.JsonConverter");
-        HERDER_CONFIG.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.copycat.json.JsonConverter");
-        HERDER_CONFIG.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.copycat.json.JsonConverter");
-    }
-    private static final String MEMBER_URL = "memberUrl";
-
-    private static final String CONN1 = "sourceA";
-    private static final String CONN2 = "sourceB";
-    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> CONN1_CONFIG = new HashMap<>();
-    static {
-        CONN1_CONFIG.put(ConnectorConfig.NAME_CONFIG, CONN1);
-        CONN1_CONFIG.put(ConnectorConfig.TASKS_MAX_CONFIG, MAX_TASKS.toString());
-        CONN1_CONFIG.put(ConnectorConfig.TOPICS_CONFIG, "foo,bar");
-        CONN1_CONFIG.put(ConnectorConfig.CONNECTOR_CLASS_CONFIG, BogusSourceConnector.class.getName());
-    }
-    private static final Map<String, String> CONN1_CONFIG_UPDATED = new HashMap<>(CONN1_CONFIG);
-    static {
-        CONN1_CONFIG_UPDATED.put(ConnectorConfig.TOPICS_CONFIG, "foo,bar,baz");
-    }
-    private static final Map<String, String> CONN2_CONFIG = new HashMap<>();
-    static {
-        CONN2_CONFIG.put(ConnectorConfig.NAME_CONFIG, CONN2);
-        CONN2_CONFIG.put(ConnectorConfig.TASKS_MAX_CONFIG, MAX_TASKS.toString());
-        CONN2_CONFIG.put(ConnectorConfig.TOPICS_CONFIG, "foo,bar");
-        CONN2_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 List<Map<String, String>> TASK_CONFIGS = new ArrayList<>();
-    static {
-        TASK_CONFIGS.add(TASK_CONFIG);
-        TASK_CONFIGS.add(TASK_CONFIG);
-        TASK_CONFIGS.add(TASK_CONFIG);
-    }
-    private static final HashMap<ConnectorTaskId, Map<String, String>> TASK_CONFIGS_MAP = new HashMap<>();
-    static {
-        TASK_CONFIGS_MAP.put(TASK0, TASK_CONFIG);
-        TASK_CONFIGS_MAP.put(TASK1, TASK_CONFIG);
-        TASK_CONFIGS_MAP.put(TASK2, TASK_CONFIG);
-    }
-    private static final ClusterConfigState SNAPSHOT = new ClusterConfigState(1, Collections.singletonMap(CONN1, 3),
-            Collections.singletonMap(CONN1, CONN1_CONFIG), TASK_CONFIGS_MAP, Collections.<String>emptySet());
-    private static final ClusterConfigState SNAPSHOT_UPDATED_CONN1_CONFIG = new ClusterConfigState(1, Collections.singletonMap(CONN1, 3),
-            Collections.singletonMap(CONN1, CONN1_CONFIG_UPDATED), TASK_CONFIGS_MAP, Collections.<String>emptySet());
-
-    @Mock private KafkaConfigStorage configStorage;
-    @Mock private WorkerGroupMember member;
-    private MockTime time;
-    private DistributedHerder herder;
-    @Mock private Worker worker;
-    @Mock private Callback<Herder.Created<ConnectorInfo>> putConnectorCallback;
-
-    private Callback<String> connectorConfigCallback;
-    private Callback<List<ConnectorTaskId>> taskConfigCallback;
-    private WorkerRebalanceListener rebalanceListener;
-
-    @Before
-    public void setUp() throws Exception {
-        worker = PowerMock.createMock(Worker.class);
-        time = new MockTime();
-
-        herder = PowerMock.createPartialMock(DistributedHerder.class, new String[]{"backoff"},
-                new DistributedConfig(HERDER_CONFIG), worker, configStorage, member, MEMBER_URL, time);
-        connectorConfigCallback = Whitebox.invokeMethod(herder, "connectorConfigCallback");
-        taskConfigCallback = Whitebox.invokeMethod(herder, "taskConfigCallback");
-        rebalanceListener = Whitebox.invokeMethod(herder, "rebalanceListener");
-    }
-
-    @Test
-    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.connectorTaskConfigs(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();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    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();
-
-        PowerMock.replayAll();
-
-        herder.halt();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCreateConnector() throws Exception {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        expectRebalance(1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
-        expectPostRebalanceCatchup(SNAPSHOT);
-
-        member.wakeup();
-        PowerMock.expectLastCall();
-        // CONN2 is new, should succeed
-        configStorage.putConnectorConfig(CONN2, CONN2_CONFIG);
-        PowerMock.expectLastCall();
-        ConnectorInfo info = new ConnectorInfo(CONN2, CONN2_CONFIG, Collections.<ConnectorTaskId>emptyList());
-        putConnectorCallback.onCompletion(null, new Herder.Created<>(true, info));
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        // No immediate action besides this -- change will be picked up via the config log
-
-        PowerMock.replayAll();
-
-        herder.putConnectorConfig(CONN2, CONN2_CONFIG, false, putConnectorCallback);
-        herder.tick();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testCreateConnectorAlreadyExists() throws Exception {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        expectRebalance(1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
-        expectPostRebalanceCatchup(SNAPSHOT);
-
-        member.wakeup();
-        PowerMock.expectLastCall();
-        // CONN1 already exists
-        putConnectorCallback.onCompletion(EasyMock.<AlreadyExistsException>anyObject(), EasyMock.<Herder.Created<ConnectorInfo>>isNull());
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        // No immediate action besides this -- change will be picked up via the config log
-
-        PowerMock.replayAll();
-
-        herder.putConnectorConfig(CONN1, CONN1_CONFIG, false, putConnectorCallback);
-        herder.tick();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    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.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
-
-        // And delete the connector
-        member.wakeup();
-        PowerMock.expectLastCall();
-        configStorage.putConnectorConfig(CONN1, null);
-        PowerMock.expectLastCall();
-        putConnectorCallback.onCompletion(null, new Herder.Created<ConnectorInfo>(false, null));
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-        // No immediate action besides this -- change will be picked up via the config log
-
-        PowerMock.replayAll();
-
-        herder.putConnectorConfig(CONN1, null, true, putConnectorCallback);
-        herder.tick();
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    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.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        herder.tick(); // join
-        connectorConfigCallback.onCompletion(null, CONN1); // read updated config
-        herder.tick(); // apply config
-        herder.tick(); // do rebalance
-
-        PowerMock.verifyAll();
-    }
-
-    @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();
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // 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.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        herder.tick(); // join
-        connectorConfigCallback.onCompletion(null, CONN1); // read updated config
-        herder.tick(); // apply config
-
-        PowerMock.verifyAll();
-    }
-
-    @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();
-
-        // 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, Collections.<String>emptyList(), Arrays.asList(TASK0));
-        worker.addTask(EasyMock.eq(TASK0), EasyMock.<TaskConfig>anyObject());
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        herder.tick(); // join
-        taskConfigCallback.onCompletion(null, Arrays.asList(TASK0, TASK1, TASK2)); // read updated config
-        herder.tick(); // apply config
-        herder.tick(); // do rebalance
-
-        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", DistributedConfig.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.connectorTaskConfigs(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();
-    }
-
-    @Test
-    public void testAccessors() throws Exception {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        expectRebalance(1, Collections.<String>emptyList(), Collections.<ConnectorTaskId>emptyList());
-        expectPostRebalanceCatchup(SNAPSHOT);
-
-
-        member.wakeup();
-        PowerMock.expectLastCall().anyTimes();
-        // list connectors, get connector info, get connector config, get task configs
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-
-        PowerMock.replayAll();
-
-        FutureCallback<Collection<String>> listConnectorsCb = new FutureCallback<>();
-        herder.connectors(listConnectorsCb);
-        FutureCallback<ConnectorInfo> connectorInfoCb = new FutureCallback<>();
-        herder.connectorInfo(CONN1, connectorInfoCb);
-        FutureCallback<Map<String, String>> connectorConfigCb = new FutureCallback<>();
-        herder.connectorConfig(CONN1, connectorConfigCb);
-        FutureCallback<List<TaskInfo>> taskConfigsCb = new FutureCallback<>();
-        herder.taskConfigs(CONN1, taskConfigsCb);
-
-        herder.tick();
-        assertTrue(listConnectorsCb.isDone());
-        assertEquals(Collections.singleton(CONN1), listConnectorsCb.get());
-        assertTrue(connectorInfoCb.isDone());
-        ConnectorInfo info = new ConnectorInfo(CONN1, CONN1_CONFIG, Arrays.asList(TASK0, TASK1, TASK2));
-        assertEquals(info, connectorInfoCb.get());
-        assertTrue(connectorConfigCb.isDone());
-        assertEquals(CONN1_CONFIG, connectorConfigCb.get());
-        assertTrue(taskConfigsCb.isDone());
-        assertEquals(Arrays.asList(
-                        new TaskInfo(TASK0, TASK_CONFIG),
-                        new TaskInfo(TASK1, TASK_CONFIG),
-                        new TaskInfo(TASK2, TASK_CONFIG)),
-                taskConfigsCb.get());
-
-        PowerMock.verifyAll();
-    }
-
-    @Test
-    public void testPutConnectorConfig() throws Exception {
-        EasyMock.expect(member.memberId()).andStubReturn("leader");
-        expectRebalance(1, Arrays.asList(CONN1), Collections.<ConnectorTaskId>emptyList());
-        expectPostRebalanceCatchup(SNAPSHOT);
-        worker.addConnector(EasyMock.<ConnectorConfig>anyObject(), EasyMock.<ConnectorContext>anyObject());
-        PowerMock.expectLastCall();
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
-
-        // list connectors, get connector info, get connector config, get task configs
-        member.wakeup();
-        PowerMock.expectLastCall().anyTimes();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // Poll loop for second round of calls
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        configStorage.putConnectorConfig(CONN1, CONN1_CONFIG_UPDATED);
-        PowerMock.expectLastCall().andAnswer(new IAnswer<Object>() {
-            @Override
-            public Object answer() throws Throwable {
-                // Simulate response to writing config + waiting until end of log to be read
-                connectorConfigCallback.onCompletion(null, CONN1);
-                return null;
-            }
-        });
-        // As a result of reconfig, should need to update snapshot. With only connector updates, we'll just restart
-        // connector without rebalance
-        EasyMock.expect(configStorage.snapshot()).andReturn(SNAPSHOT_UPDATED_CONN1_CONFIG);
-        worker.stopConnector(CONN1);
-        PowerMock.expectLastCall();
-        Capture<ConnectorConfig> capturedUpdatedConfig = EasyMock.newCapture();
-        worker.addConnector(EasyMock.capture(capturedUpdatedConfig), EasyMock.<ConnectorContext>anyObject());
-        PowerMock.expectLastCall();
-        EasyMock.expect(worker.connectorTaskConfigs(CONN1, MAX_TASKS, null)).andReturn(TASK_CONFIGS);
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        // Third tick just to read the config
-        member.ensureActive();
-        PowerMock.expectLastCall();
-        member.poll(EasyMock.anyInt());
-        PowerMock.expectLastCall();
-
-        PowerMock.replayAll();
-
-        // Should pick up original config
-        FutureCallback<Map<String, String>> connectorConfigCb = new FutureCallback<>();
-        herder.connectorConfig(CONN1, connectorConfigCb);
-        herder.tick();
-        assertTrue(connectorConfigCb.isDone());
-        assertEquals(CONN1_CONFIG, connectorConfigCb.get());
-
-        // Apply new config.
-        FutureCallback<Herder.Created<ConnectorInfo>> putConfigCb = new FutureCallback<>();
-        herder.putConnectorConfig(CONN1, CONN1_CONFIG_UPDATED, true, putConfigCb);
-        herder.tick();
-        assertTrue(putConfigCb.isDone());
-        ConnectorInfo updatedInfo = new ConnectorInfo(CONN1, CONN1_CONFIG_UPDATED, Arrays.asList(TASK0, TASK1, TASK2));
-        assertEquals(new Herder.Created<>(false, updatedInfo), putConfigCb.get());
-
-        // Check config again to validate change
-        connectorConfigCb = new FutureCallback<>();
-        herder.connectorConfig(CONN1, connectorConfigCb);
-        herder.tick();
-        assertTrue(connectorConfigCb.isDone());
-        assertEquals(CONN1_CONFIG_UPDATED, connectorConfigCb.get());
-        // The config passed to Worker should
-        assertEquals(Arrays.asList("foo", "bar", "baz"),
-                capturedUpdatedConfig.getValue().getList(ConnectorConfig.TOPICS_CONFIG));
-        PowerMock.verifyAll();
-    }
-
-    @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 expectRebalance(final long offset, final List<String> assignedConnectors, final List<ConnectorTaskId> assignedTasks) {
-        expectRebalance(null, null, CopycatProtocol.Assignment.NO_ERROR, offset, assignedConnectors, assignedTasks);
-    }
-
-    // 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", "leaderUrl", offset, assignedConnectors, assignedTasks);
-                rebalanceListener.onAssigned(assignment);
-                return null;
-            }
-        });
-        member.wakeup();
-        PowerMock.expectLastCall();
-    }
-
-    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);
-    }
-
-
-    // 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 BogusSourceTask extends SourceTask {
-    }
-
-}


Mime
View raw message