[14/26] kafka git commit: KAFKA-2774: Rename Copycat to Kafka Connect
http://git-wip-us.apache.org/repos/asf/kafka/blob/417e283d/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
--
diff --git
a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
new file mode 100644
index 000..4e54bf1
--- /dev/null
+++
b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
@@ -0,0 +1,357 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+package org.apache.kafka.connect.storage;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.util.Callback;
+import org.apache.kafka.connect.util.KafkaBasedLog;
+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.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(KafkaOffsetBackingStore.class)
+@PowerMockIgnore("javax.management.*")
+public class KafkaOffsetBackingStoreTest {
+private static final String TOPIC = "connect-offsets";
+private static final Map DEFAULT_PROPS = new HashMap<>();
+static {
+DEFAULT_PROPS.put(KafkaOffsetBackingStore.OFFSET_STORAGE_TOPIC_CONFIG,
TOPIC);
+DEFAULT_PROPS.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
"broker1:9092,broker2:9093");
+}
+private static final Map FIRST_SET = new
HashMap<>();
+static {
+FIRST_SET.put(buffer("key"), buffer("value"));
+FIRST_SET.put(null, null);
+}
+
+private static final ByteBuffer TP0_KEY = buffer("TP0KEY");
+private static final ByteBuffer TP1_KEY = buffer("TP1KEY");
+private static final ByteBuffer TP2_KEY = buffer("TP2KEY");
+private static final ByteBuffer TP0_VALUE = buffer("VAL0");
+private static final ByteBuffer TP1_VALUE = buffer("VAL1");
+private static final ByteBuffer TP2_VALUE = buffer("VAL2");
+private static final ByteBuffer TP0_VALUE_NEW = buffer("VAL0_NEW");
+private static final ByteBuffer TP1_VALUE_NEW = buffer("VAL1_NEW");
+
+@Mock
+KafkaBasedLog storeLog;
+private KafkaOffsetBackingStore store;
+
+private Capture capturedTopic = EasyMock.newCapture();
+private Capture> capturedProducerProps =
EasyMock.newCapture();
+private Capture> capturedConsumerProps =
EasyMock.newCapture();
+private Capture>>
capturedConsumedCallback = EasyMock.newCapture();
+
+@Before
+public void setUp() throws Exception {
+store =
PowerMock.createPartialMockAndInvokeDefaultConstructor(KafkaOffsetBackingStore.class,
new String[]{"createKafkaBasedLog"});
+}
+
+@Test(expected = ConnectException.class)
+public void testMissingTopic() {
+
[14/26] kafka git commit: KAFKA-2774: Rename Copycat to Kafka Connect
http://git-wip-us.apache.org/repos/asf/kafka/blob/f2031d40/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
--
diff --git
a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
new file mode 100644
index 000..4e54bf1
--- /dev/null
+++
b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
@@ -0,0 +1,357 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ **/
+
+package org.apache.kafka.connect.storage;
+
+import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.util.Callback;
+import org.apache.kafka.connect.util.KafkaBasedLog;
+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.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest(KafkaOffsetBackingStore.class)
+@PowerMockIgnore("javax.management.*")
+public class KafkaOffsetBackingStoreTest {
+private static final String TOPIC = "connect-offsets";
+private static final Map DEFAULT_PROPS = new HashMap<>();
+static {
+DEFAULT_PROPS.put(KafkaOffsetBackingStore.OFFSET_STORAGE_TOPIC_CONFIG,
TOPIC);
+DEFAULT_PROPS.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
"broker1:9092,broker2:9093");
+}
+private static final Map FIRST_SET = new
HashMap<>();
+static {
+FIRST_SET.put(buffer("key"), buffer("value"));
+FIRST_SET.put(null, null);
+}
+
+private static final ByteBuffer TP0_KEY = buffer("TP0KEY");
+private static final ByteBuffer TP1_KEY = buffer("TP1KEY");
+private static final ByteBuffer TP2_KEY = buffer("TP2KEY");
+private static final ByteBuffer TP0_VALUE = buffer("VAL0");
+private static final ByteBuffer TP1_VALUE = buffer("VAL1");
+private static final ByteBuffer TP2_VALUE = buffer("VAL2");
+private static final ByteBuffer TP0_VALUE_NEW = buffer("VAL0_NEW");
+private static final ByteBuffer TP1_VALUE_NEW = buffer("VAL1_NEW");
+
+@Mock
+KafkaBasedLog storeLog;
+private KafkaOffsetBackingStore store;
+
+private Capture capturedTopic = EasyMock.newCapture();
+private Capture> capturedProducerProps =
EasyMock.newCapture();
+private Capture> capturedConsumerProps =
EasyMock.newCapture();
+private Capture>>
capturedConsumedCallback = EasyMock.newCapture();
+
+@Before
+public void setUp() throws Exception {
+store =
PowerMock.createPartialMockAndInvokeDefaultConstructor(KafkaOffsetBackingStore.class,
new String[]{"createKafkaBasedLog"});
+}
+
+@Test(expected = ConnectException.class)
+public void testMissingTopic() {
+
