diff --git a/bin/copycat-distributed.sh b/bin/copycat-distributed.sh new file mode 100755 index 0000000000000..4d62300e00d13 --- /dev/null +++ b/bin/copycat-distributed.sh @@ -0,0 +1,23 @@ +#!/bin/sh +# 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. + +base_dir=$(dirname $0) + +if [ "x$KAFKA_LOG4J_OPTS" = "x" ]; then + export KAFKA_LOG4J_OPTS="-Dlog4j.configuration=file:$base_dir/../config/copycat-log4j.properties" +fi + +exec $(dirname $0)/kafka-run-class.sh org.apache.kafka.copycat.cli.CopycatDistributed "$@" diff --git a/checkstyle/import-control.xml b/checkstyle/import-control.xml index f24977b5020ab..d58c472872ced 100644 --- a/checkstyle/import-control.xml +++ b/checkstyle/import-control.xml @@ -27,6 +27,9 @@ + + + @@ -124,6 +127,7 @@ + @@ -138,14 +142,11 @@ - - - - + @@ -153,9 +154,6 @@ - - - diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java index d2dcbe3231f5f..3009f6b6f01d3 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java @@ -1023,10 +1023,9 @@ public long position(TopicPartition partition) { Long offset = this.subscriptions.consumed(partition); if (offset == null) { updateFetchPositions(Collections.singleton(partition)); - return this.subscriptions.consumed(partition); - } else { - return offset; + offset = this.subscriptions.consumed(partition); } + return offset; } finally { release(); } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java index e7e36189de294..1f802a8097072 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java @@ -14,26 +14,32 @@ import org.apache.kafka.clients.consumer.internals.NoOpConsumerRebalanceListener; import org.apache.kafka.clients.consumer.internals.SubscriptionState; +import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.Metric; import org.apache.kafka.common.MetricName; import org.apache.kafka.common.PartitionInfo; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.TimeoutException; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Map.Entry; import java.util.Set; import java.util.regex.Pattern; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; /** * A mock of the {@link Consumer} interface you can use for testing code that uses Kafka. This class is not - * threadsafe - *

- * The consumer runs in the user thread and multiplexes I/O over TCP connections to each of the brokers it needs to - * communicate with. Failure to close the consumer after use will leak these resources. + * threadsafe . However, you can use the {@link #waitForPollThen(Runnable,long)} method to write multithreaded tests + * where a driver thread waits for {@link #poll(long)} to be called and then can safely perform operations during a + * callback. */ public class MockConsumer implements Consumer { @@ -41,26 +47,45 @@ public class MockConsumer implements Consumer { private final SubscriptionState subscriptions; private Map>> records; private boolean closed; + private final Map beginningOffsets; + private final Map endOffsets; + + private AtomicReference pollLatch; + private KafkaException exception; + + private AtomicBoolean wakeup; public MockConsumer(OffsetResetStrategy offsetResetStrategy) { this.subscriptions = new SubscriptionState(offsetResetStrategy); this.partitions = new HashMap>(); this.records = new HashMap>>(); this.closed = false; + this.beginningOffsets = new HashMap<>(); + this.endOffsets = new HashMap<>(); + this.pollLatch = new AtomicReference<>(); + this.exception = null; + this.wakeup = new AtomicBoolean(false); } @Override - public synchronized Set assignment() { + public Set assignment() { return this.subscriptions.assignedPartitions(); } + /** Simulate a rebalance event. */ + public void rebalance(Collection newAssignment) { + // TODO: Rebalance callbacks + this.records.clear(); + this.subscriptions.changePartitionAssignment(newAssignment); + } + @Override - public synchronized Set subscription() { + public Set subscription() { return this.subscriptions.subscription(); } @Override - public synchronized void subscribe(List topics) { + public void subscribe(List topics) { subscribe(topics, new NoOpConsumerRebalanceListener()); } @@ -79,13 +104,13 @@ public void subscribe(Pattern pattern, final ConsumerRebalanceListener listener) } @Override - public synchronized void subscribe(List topics, final ConsumerRebalanceListener listener) { + public void subscribe(List topics, final ConsumerRebalanceListener listener) { ensureNotClosed(); this.subscriptions.subscribe(topics, listener); } @Override - public synchronized void assign(List partitions) { + public void assign(List partitions) { ensureNotClosed(); this.subscriptions.assign(partitions); } @@ -97,14 +122,39 @@ public void unsubscribe() { } @Override - public synchronized ConsumerRecords poll(long timeout) { + public ConsumerRecords poll(long timeout) { ensureNotClosed(); + + CountDownLatch pollLatchCopy = pollLatch.get(); + if (pollLatchCopy != null) { + pollLatch.set(null); + pollLatchCopy.countDown(); + synchronized (pollLatchCopy) { + // Will block until caller of waitUntilPollThen() finishes their callback. + } + } + + if (wakeup.get()) { + wakeup.set(false); + throw new ConsumerWakeupException(); + } + + if (exception != null) { + RuntimeException exception = this.exception; + this.exception = null; + throw exception; + } + + // Handle seeks that need to wait for a poll() call to be processed + for (TopicPartition tp : subscriptions.missingFetchPositions()) + updateFetchPosition(tp); + // update the consumed offset for (Map.Entry>> entry : this.records.entrySet()) { if (!subscriptions.isPaused(entry.getKey())) { List> recs = entry.getValue(); if (!recs.isEmpty()) - this.subscriptions.consumed(entry.getKey(), recs.get(recs.size() - 1).offset()); + this.subscriptions.consumed(entry.getKey(), recs.get(recs.size() - 1).offset() + 1); } } @@ -113,15 +163,12 @@ public synchronized ConsumerRecords poll(long timeout) { return copy; } - public synchronized void addRecord(ConsumerRecord record) { + public void addRecord(ConsumerRecord record) { ensureNotClosed(); TopicPartition tp = new TopicPartition(record.topic(), record.partition()); - ArrayList currentAssigned = new ArrayList<>(this.subscriptions.assignedPartitions()); - if (!currentAssigned.contains(tp)) { - currentAssigned.add(tp); - this.subscriptions.changePartitionAssignment(currentAssigned); - } - subscriptions.seek(tp, record.offset()); + Set currentAssigned = new HashSet<>(this.subscriptions.assignedPartitions()); + if (!currentAssigned.contains(tp)) + throw new IllegalStateException("Cannot add records for a partition that is not assigned to the consumer"); List> recs = this.records.get(tp); if (recs == null) { recs = new ArrayList>(); @@ -130,10 +177,14 @@ public synchronized void addRecord(ConsumerRecord record) { recs.add(record); } + public void setException(KafkaException exception) { + this.exception = exception; + } + @Override - public synchronized void commitAsync(Map offsets, OffsetCommitCallback callback) { + public void commitAsync(Map offsets, OffsetCommitCallback callback) { ensureNotClosed(); - for (Entry entry : offsets.entrySet()) + for (Map.Entry entry : offsets.entrySet()) subscriptions.committed(entry.getKey(), entry.getValue()); if (callback != null) { callback.onComplete(offsets, null); @@ -141,54 +192,71 @@ public synchronized void commitAsync(Map offs } @Override - public synchronized void commitSync(Map offsets) { + public void commitSync(Map offsets) { commitAsync(offsets, null); } @Override - public synchronized void commitAsync() { + public void commitAsync() { commitAsync(null); } @Override - public synchronized void commitAsync(OffsetCommitCallback callback) { + public void commitAsync(OffsetCommitCallback callback) { ensureNotClosed(); commitAsync(this.subscriptions.allConsumed(), callback); } @Override - public synchronized void commitSync() { + public void commitSync() { commitSync(this.subscriptions.allConsumed()); } @Override - public synchronized void seek(TopicPartition partition, long offset) { + public void seek(TopicPartition partition, long offset) { ensureNotClosed(); subscriptions.seek(partition, offset); } @Override - public synchronized OffsetAndMetadata committed(TopicPartition partition) { + public OffsetAndMetadata committed(TopicPartition partition) { ensureNotClosed(); return subscriptions.committed(partition); } @Override - public synchronized long position(TopicPartition partition) { + public long position(TopicPartition partition) { ensureNotClosed(); - return subscriptions.consumed(partition); + if (!this.subscriptions.isAssigned(partition)) + throw new IllegalArgumentException("You can only check the position for partitions assigned to this consumer."); + Long offset = this.subscriptions.consumed(partition); + if (offset == null) { + updateFetchPosition(partition); + offset = this.subscriptions.consumed(partition); + } + return offset; } @Override - public synchronized void seekToBeginning(TopicPartition... partitions) { + public void seekToBeginning(TopicPartition... partitions) { ensureNotClosed(); - throw new UnsupportedOperationException(); + for (TopicPartition tp : partitions) + subscriptions.needOffsetReset(tp, OffsetResetStrategy.EARLIEST); + } + + public void updateBeginningOffsets(Map newOffsets) { + beginningOffsets.putAll(newOffsets); } @Override - public synchronized void seekToEnd(TopicPartition... partitions) { + public void seekToEnd(TopicPartition... partitions) { ensureNotClosed(); - throw new UnsupportedOperationException(); + for (TopicPartition tp : partitions) + subscriptions.needOffsetReset(tp, OffsetResetStrategy.LATEST); + } + + public void updateEndOffsets(Map newOffsets) { + endOffsets.putAll(newOffsets); } @Override @@ -198,7 +266,7 @@ public synchronized void seekToEnd(TopicPartition... partitions) { } @Override - public synchronized List partitionsFor(String topic) { + public List partitionsFor(String topic) { ensureNotClosed(); List parts = this.partitions.get(topic); if (parts == null) @@ -213,7 +281,7 @@ public Map> listTopics() { return partitions; } - public synchronized void updatePartitions(String topic, List partitions) { + public void updatePartitions(String topic, List partitions) { ensureNotClosed(); this.partitions.put(topic, partitions); } @@ -231,18 +299,69 @@ public void resume(TopicPartition... partitions) { } @Override - public synchronized void close() { + public void close() { ensureNotClosed(); this.closed = true; } + public boolean closed() { + return this.closed; + } + @Override public void wakeup() { + wakeup.set(true); + } + public void waitForPoll(long timeoutMs) { + waitForPollThen(null, timeoutMs); + } + + public void waitForPollThen(Runnable task, long timeoutMs) { + CountDownLatch latch = new CountDownLatch(1); + synchronized (latch) { + pollLatch.set(latch); + try { + if (!latch.await(timeoutMs, TimeUnit.MILLISECONDS)) + throw new TimeoutException("Timed out waiting for consumer thread to call poll()."); + } catch (InterruptedException e) { + throw new IllegalStateException("MockConsumer waiting thread was interrupted.", e); + } + if (task != null) + task.run(); + } } private void ensureNotClosed() { if (this.closed) throw new IllegalStateException("This consumer has already been closed."); } + + private void updateFetchPosition(TopicPartition tp) { + if (subscriptions.isOffsetResetNeeded(tp)) { + resetOffsetPosition(tp); + } else if (subscriptions.committed(tp) == null) { + subscriptions.needOffsetReset(tp); + resetOffsetPosition(tp); + } else { + subscriptions.seek(tp, subscriptions.committed(tp).offset()); + } + } + + private void resetOffsetPosition(TopicPartition tp) { + OffsetResetStrategy strategy = subscriptions.resetStrategy(tp); + Long offset; + if (strategy == OffsetResetStrategy.EARLIEST) { + offset = beginningOffsets.get(tp); + if (offset == null) + throw new IllegalStateException("MockConsumer didn't have beginning offset specified, but tried to seek to beginning"); + } else if (strategy == OffsetResetStrategy.LATEST) { + offset = endOffsets.get(tp); + if (offset == null) + throw new IllegalStateException("MockConsumer didn't have end offset specified, but tried to seek to end"); + } else { + throw new NoOffsetForPartitionException("No offset available"); + } + seek(tp, offset); + } } diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java index 9b610d87f27a2..25a0e9071f5d6 100644 --- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java +++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java @@ -17,6 +17,7 @@ import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; +import java.util.Collection; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; @@ -264,7 +265,7 @@ public boolean hasAllFetchPositions() { } public Set missingFetchPositions() { - Set missing = new HashSet<>(this.assignment.keySet()); + Set missing = new HashSet<>(); for (Map.Entry entry : assignment.entrySet()) if (!entry.getValue().hasValidPosition) missing.add(entry.getKey()); @@ -275,7 +276,7 @@ public boolean partitionAssignmentNeeded() { return this.needsPartitionAssignment; } - public void changePartitionAssignment(List assignments) { + public void changePartitionAssignment(Collection assignments) { for (TopicPartition tp : assignments) if (!this.subscription.contains(tp.topic())) throw new IllegalArgumentException("Assigned partition " + tp + " for non-subscribed topic."); diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java index f702535ca115d..fa06be9e2ba2a 100644 --- a/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/consumer/MockConsumerTest.java @@ -21,6 +21,7 @@ import org.junit.Test; import java.util.Arrays; +import java.util.HashMap; import java.util.Iterator; import static org.junit.Assert.assertEquals; @@ -34,6 +35,13 @@ public class MockConsumerTest { public void testSimpleMock() { consumer.subscribe(Arrays.asList("test"), new NoOpConsumerRebalanceListener()); assertEquals(0, consumer.poll(1000).count()); + consumer.rebalance(Arrays.asList(new TopicPartition("test", 0), new TopicPartition("test", 1))); + // Mock consumers need to seek manually since they cannot automatically reset offsets + HashMap beginningOffsets = new HashMap<>(); + beginningOffsets.put(new TopicPartition("test", 0), 0L); + beginningOffsets.put(new TopicPartition("test", 1), 0L); + consumer.updateBeginningOffsets(beginningOffsets); + consumer.seek(new TopicPartition("test", 0), 0); ConsumerRecord rec1 = new ConsumerRecord("test", 0, 0, "key1", "value1"); ConsumerRecord rec2 = new ConsumerRecord("test", 0, 1, "key2", "value2"); consumer.addRecord(rec1); @@ -43,9 +51,9 @@ public void testSimpleMock() { assertEquals(rec1, iter.next()); assertEquals(rec2, iter.next()); assertFalse(iter.hasNext()); - assertEquals(1L, consumer.position(new TopicPartition("test", 0))); + assertEquals(2L, consumer.position(new TopicPartition("test", 0))); consumer.commitSync(); - assertEquals(1L, consumer.committed(new TopicPartition("test", 0)).offset()); + assertEquals(2L, consumer.committed(new TopicPartition("test", 0)).offset()); } } diff --git a/config/copycat-distributed.properties b/config/copycat-distributed.properties new file mode 100644 index 0000000000000..654ed24dad2ef --- /dev/null +++ b/config/copycat-distributed.properties @@ -0,0 +1,39 @@ +## +# 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. +## + +# These are defaults. This file just demonstrates how to override some settings. +bootstrap.servers=localhost:9092 + +# The converters specify the format of data in Kafka and how to translate it into Copycat data. Every Copycat user will +# need to configure these based on the format they want their data in when loaded from or stored into Kafka +key.converter=org.apache.kafka.copycat.json.JsonConverter +value.converter=org.apache.kafka.copycat.json.JsonConverter +# Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply +# it to +key.converter.schemas.enable=true +value.converter.schemas.enable=true + +# The offset converter is configurable and must be specified, but most users will always want to use the built-in default. +# Offset data is never visible outside of Copcyat. +offset.key.converter=org.apache.kafka.copycat.json.JsonConverter +offset.value.converter=org.apache.kafka.copycat.json.JsonConverter +offset.key.converter.schemas.enable=false +offset.value.converter.schemas.enable=false + +offset.storage.topic=copycat-offsets +# Flush much faster than normal, which is useful for testing/debugging +offset.flush.interval.ms=10000 diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java new file mode 100644 index 0000000000000..b5e88964b559e --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatDistributed.java @@ -0,0 +1,87 @@ +/** + * 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.cli; + +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.copycat.runtime.Copycat; +import org.apache.kafka.copycat.runtime.Herder; +import org.apache.kafka.copycat.runtime.Worker; +import org.apache.kafka.copycat.runtime.standalone.StandaloneHerder; +import org.apache.kafka.copycat.storage.KafkaOffsetBackingStore; +import org.apache.kafka.copycat.util.Callback; +import org.apache.kafka.copycat.util.FutureCallback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.Properties; + +/** + *

+ * Command line utility that runs Copycat in distributed mode. In this mode, the process joints a group of other workers + * and work is distributed among them. This is useful for running Copycat as a service, where connectors can be + * submitted to the cluster to be automatically executed in a scalable, distributed fashion. This also allows you to + * easily scale out horizontally, elastically adding or removing capacity simply by starting or stopping worker + * instances. + *

+ */ +@InterfaceStability.Unstable +public class CopycatDistributed { + private static final Logger log = LoggerFactory.getLogger(CopycatDistributed.class); + + public static void main(String[] args) throws Exception { + Properties workerProps; + Properties connectorProps; + + if (args.length < 2) { + log.info("Usage: CopycatDistributed worker.properties connector1.properties [connector2.properties ...]"); + System.exit(1); + } + + String workerPropsFile = args[0]; + workerProps = !workerPropsFile.isEmpty() ? Utils.loadProps(workerPropsFile) : new Properties(); + + WorkerConfig workerConfig = new WorkerConfig(workerProps); + Worker worker = new Worker(workerConfig, new KafkaOffsetBackingStore()); + Herder herder = new StandaloneHerder(worker); + final Copycat copycat = new Copycat(worker, herder); + copycat.start(); + + try { + for (final String connectorPropsFile : Arrays.copyOfRange(args, 1, args.length)) { + connectorProps = Utils.loadProps(connectorPropsFile); + FutureCallback cb = new FutureCallback<>(new Callback() { + @Override + public void onCompletion(Throwable error, String id) { + if (error != null) + log.error("Failed to create job for {}", connectorPropsFile); + } + }); + herder.addConnector(connectorProps, cb); + cb.get(); + } + } catch (Throwable t) { + log.error("Stopping after connector error", t); + copycat.stop(); + } + + // Shutdown will be triggered by Ctrl-C or via HTTP shutdown request + copycat.awaitStop(); + } +} diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java index 130a5294c3e19..12ec154b6a8a6 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java @@ -19,9 +19,11 @@ import org.apache.kafka.common.annotation.InterfaceStability; import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.copycat.runtime.Copycat; import org.apache.kafka.copycat.runtime.Herder; import org.apache.kafka.copycat.runtime.Worker; import org.apache.kafka.copycat.runtime.standalone.StandaloneHerder; +import org.apache.kafka.copycat.storage.FileOffsetBackingStore; import org.apache.kafka.copycat.util.Callback; import org.apache.kafka.copycat.util.FutureCallback; import org.slf4j.Logger; @@ -58,9 +60,9 @@ public static void main(String[] args) throws Exception { workerProps = !workerPropsFile.isEmpty() ? Utils.loadProps(workerPropsFile) : new Properties(); WorkerConfig workerConfig = new WorkerConfig(workerProps); - Worker worker = new Worker(workerConfig); + Worker worker = new Worker(workerConfig, new FileOffsetBackingStore()); Herder herder = new StandaloneHerder(worker); - final org.apache.kafka.copycat.runtime.Copycat copycat = new org.apache.kafka.copycat.runtime.Copycat(worker, herder); + final Copycat copycat = new Copycat(worker, herder); copycat.start(); try { diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Worker.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Worker.java index 6cbce0b251e3e..a34a014ce93d1 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Worker.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Worker.java @@ -59,8 +59,8 @@ public class Worker { private KafkaProducer producer; private SourceTaskOffsetCommitter sourceTaskOffsetCommitter; - public Worker(WorkerConfig config) { - this(new SystemTime(), config, null); + public Worker(WorkerConfig config, OffsetBackingStore offsetBackingStore) { + this(new SystemTime(), config, offsetBackingStore); } @SuppressWarnings("unchecked") @@ -76,12 +76,8 @@ public Worker(Time time, WorkerConfig config, OffsetBackingStore offsetBackingSt this.offsetValueConverter = config.getConfiguredInstance(WorkerConfig.OFFSET_VALUE_CONVERTER_CLASS_CONFIG, Converter.class); this.offsetValueConverter.configure(config.originalsWithPrefix("offset.value.converter."), false); - if (offsetBackingStore != null) { - this.offsetBackingStore = offsetBackingStore; - } else { - this.offsetBackingStore = new FileOffsetBackingStore(); - this.offsetBackingStore.configure(config.originals()); - } + this.offsetBackingStore = offsetBackingStore; + this.offsetBackingStore.configure(config.originals()); } public void start() { @@ -132,7 +128,7 @@ public void stop() { long timeoutMs = limit - time.milliseconds(); sourceTaskOffsetCommitter.close(timeoutMs); - offsetBackingStore.start(); + offsetBackingStore.stop(); log.info("Worker stopped"); } diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/FileOffsetBackingStore.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/FileOffsetBackingStore.java index dfa9e7808e09d..f707fd6f53753 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/FileOffsetBackingStore.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/FileOffsetBackingStore.java @@ -68,17 +68,12 @@ private void load() { Object obj = is.readObject(); if (!(obj instanceof HashMap)) throw new CopycatException("Expected HashMap but found " + obj.getClass()); - HashMap> raw = (HashMap>) obj; + Map raw = (Map) obj; data = new HashMap<>(); - for (Map.Entry> entry : raw.entrySet()) { - HashMap converted = new HashMap<>(); - for (Map.Entry mapEntry : entry.getValue().entrySet()) { - ByteBuffer key = (mapEntry.getKey() != null) ? ByteBuffer.wrap(mapEntry.getKey()) : null; - ByteBuffer value = (mapEntry.getValue() != null) ? ByteBuffer.wrap(mapEntry.getValue()) : - null; - converted.put(key, value); - } - data.put(entry.getKey(), converted); + for (Map.Entry mapEntry : raw.entrySet()) { + ByteBuffer key = (mapEntry.getKey() != null) ? ByteBuffer.wrap(mapEntry.getKey()) : null; + ByteBuffer value = (mapEntry.getValue() != null) ? ByteBuffer.wrap(mapEntry.getValue()) : null; + data.put(key, value); } is.close(); } catch (FileNotFoundException | EOFException e) { @@ -92,15 +87,11 @@ private void load() { protected void save() { try { ObjectOutputStream os = new ObjectOutputStream(new FileOutputStream(file)); - HashMap> raw = new HashMap<>(); - for (Map.Entry> entry : data.entrySet()) { - HashMap converted = new HashMap<>(); - for (Map.Entry mapEntry : entry.getValue().entrySet()) { - byte[] key = (mapEntry.getKey() != null) ? mapEntry.getKey().array() : null; - byte[] value = (mapEntry.getValue() != null) ? mapEntry.getValue().array() : null; - converted.put(key, value); - } - raw.put(entry.getKey(), converted); + Map raw = new HashMap<>(); + for (Map.Entry mapEntry : data.entrySet()) { + byte[] key = (mapEntry.getKey() != null) ? mapEntry.getKey().array() : null; + byte[] value = (mapEntry.getValue() != null) ? mapEntry.getValue().array() : null; + raw.put(key, value); } os.writeObject(raw); os.close(); diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/KafkaOffsetBackingStore.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/KafkaOffsetBackingStore.java new file mode 100644 index 0000000000000..a74b39cfed1e1 --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/KafkaOffsetBackingStore.java @@ -0,0 +1,393 @@ +/** + * 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.storage; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.ConsumerWakeupException; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.utils.SystemTime; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Utils; +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.util.Callback; +import org.apache.kafka.copycat.util.ConvertingFutureCallback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.ByteBuffer; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +/** + *

+ * Implementation of OffsetBackingStore that uses a Kafka topic to store offset data. + *

+ *

+ * Internally, this implementation both produces to and consumes from a Kafka topic which stores the offsets. + * It accepts producer and consumer overrides via its configuration but forces some settings to specific values + * to ensure correct behavior (e.g. acks, auto.offset.reset). + *

+ */ +public class KafkaOffsetBackingStore implements OffsetBackingStore { + private static final Logger log = LoggerFactory.getLogger(KafkaOffsetBackingStore.class); + + public final static String OFFSET_STORAGE_TOPIC_CONFIG = "offset.storage.topic"; + + private final static long CREATE_TOPIC_TIMEOUT_MS = 30000; + + private Time time; + private Map configs; + private String topic; + private Consumer consumer; + private Producer producer; + private HashMap data; + + private Thread thread; + private boolean stopRequested; + private Queue> readLogEndOffsetCallbacks; + + public KafkaOffsetBackingStore() { + this(new SystemTime()); + } + + public KafkaOffsetBackingStore(Time time) { + this.time = time; + } + + @Override + public void configure(Map configs) { + this.configs = configs; + topic = (String) configs.get(OFFSET_STORAGE_TOPIC_CONFIG); + if (topic == null) + throw new CopycatException("Offset storage topic must be specified"); + + data = new HashMap<>(); + stopRequested = false; + readLogEndOffsetCallbacks = new ArrayDeque<>(); + } + + @Override + public void start() { + log.info("Starting KafkaOffsetBackingStore with topic " + topic); + + producer = createProducer(); + consumer = createConsumer(); + List partitions = new ArrayList<>(); + + // Until we have admin utilities we can use to check for the existence of this topic and create it if it is missing, + // we rely on topic auto-creation + List partitionInfos = null; + long started = time.milliseconds(); + while (partitionInfos == null && time.milliseconds() - started < CREATE_TOPIC_TIMEOUT_MS) { + partitionInfos = consumer.partitionsFor(topic); + Utils.sleep(Math.min(time.milliseconds() - started, 1000)); + } + if (partitionInfos == null) + throw new CopycatException("Could not look up partition metadata for offset backing store topic in" + + " allotted period. This could indicate a connectivity issue, unavailable topic partitions, or if" + + " this is your first use of the topic it may have taken too long to create."); + + for (PartitionInfo partition : partitionInfos) + partitions.add(new TopicPartition(partition.topic(), partition.partition())); + consumer.assign(partitions); + + readToLogEnd(); + + thread = new WorkThread(); + thread.start(); + + log.info("Finished reading offsets topic and starting KafkaOffsetBackingStore"); + } + + @Override + public void stop() { + log.info("Stopping KafkaOffsetBackingStore"); + + synchronized (this) { + stopRequested = true; + consumer.wakeup(); + } + + try { + thread.join(); + } catch (InterruptedException e) { + throw new CopycatException("Failed to stop KafkaOffsetBackingStore. Exiting without cleanly shutting " + + "down it's producer and consumer.", e); + } + + try { + producer.close(); + } catch (KafkaException e) { + log.error("Failed to close KafkaOffsetBackingStore producer", e); + } + + try { + consumer.close(); + } catch (KafkaException e) { + log.error("Failed to close KafkaOffsetBackingStore consumer", e); + } + } + + @Override + public Future> get(final Collection keys, + final Callback> callback) { + ConvertingFutureCallback> future = new ConvertingFutureCallback>(callback) { + @Override + public Map convert(Void result) { + Map values = new HashMap<>(); + synchronized (KafkaOffsetBackingStore.this) { + for (ByteBuffer key : keys) + values.put(key, data.get(key)); + } + return values; + } + }; + readLogEndOffsetCallbacks.add(future); + consumer.wakeup(); + return future; + } + + @Override + public Future set(final Map values, final Callback callback) { + SetCallbackFuture producerCallback = new SetCallbackFuture(values.size(), callback); + + for (Map.Entry entry : values.entrySet()) { + producer.send(new ProducerRecord<>(topic, entry.getKey().array(), entry.getValue().array()), producerCallback); + } + + return producerCallback; + } + + + + private Producer createProducer() { + Map producerProps = new HashMap<>(); + producerProps.putAll(configs); + producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); + producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer"); + producerProps.put(ProducerConfig.ACKS_CONFIG, "all"); + return new KafkaProducer<>(producerProps); + } + + private Consumer createConsumer() { + Map consumerConfig = new HashMap<>(); + consumerConfig.putAll(configs); + consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false); + consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); + consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer"); + return new KafkaConsumer<>(consumerConfig); + } + + private void poll(long timeoutMs) { + try { + ConsumerRecords records = consumer.poll(timeoutMs); + for (ConsumerRecord record : records) { + ByteBuffer key = record.key() != null ? ByteBuffer.wrap((byte[]) record.key()) : null; + ByteBuffer value = record.value() != null ? ByteBuffer.wrap((byte[]) record.value()) : null; + data.put(key, value); + } + } catch (ConsumerWakeupException e) { + // Expected on get() or stop(). The calling code should handle this + throw e; + } catch (KafkaException e) { + log.error("Error polling: " + e); + } + } + + private void readToLogEnd() { + log.trace("Reading to end of offset log"); + + Set assignment = consumer.assignment(); + + // This approach to getting the current end offset is hacky until we have an API for looking these up directly + Map offsets = new HashMap<>(); + for (TopicPartition tp : assignment) { + long offset = consumer.position(tp); + offsets.put(tp, offset); + consumer.seekToEnd(tp); + } + + Map endOffsets = new HashMap<>(); + try { + poll(0); + } finally { + // If there is an exception, even a possibly expected one like ConsumerWakeupException, we need to make sure + // the consumers position is reset or it'll get into an inconsistent state. + for (TopicPartition tp : assignment) { + long startOffset = offsets.get(tp); + long endOffset = consumer.position(tp); + if (endOffset > startOffset) { + endOffsets.put(tp, endOffset); + consumer.seek(tp, startOffset); + } + log.trace("Reading to end of log for {}: starting offset {} to ending offset {}", tp, startOffset, endOffset); + } + } + + while (!endOffsets.isEmpty()) { + poll(Integer.MAX_VALUE); + + Iterator> it = endOffsets.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry entry = it.next(); + if (consumer.position(entry.getKey()) >= entry.getValue()) + it.remove(); + else + break; + } + } + } + + + private static class SetCallbackFuture implements org.apache.kafka.clients.producer.Callback, Future { + private int numLeft; + private boolean completed = false; + private Throwable exception = null; + private final Callback callback; + + public SetCallbackFuture(int numRecords, Callback callback) { + numLeft = numRecords; + this.callback = callback; + } + + @Override + public synchronized void onCompletion(RecordMetadata metadata, Exception exception) { + if (exception != null) { + if (!completed) { + this.exception = exception; + callback.onCompletion(exception, null); + completed = true; + this.notify(); + } + return; + } + + numLeft -= 1; + if (numLeft == 0) { + callback.onCompletion(null, null); + completed = true; + this.notify(); + } + } + + @Override + public synchronized boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + + @Override + public synchronized boolean isCancelled() { + return false; + } + + @Override + public synchronized boolean isDone() { + return completed; + } + + @Override + public synchronized Void get() throws InterruptedException, ExecutionException { + while (!completed) { + this.wait(); + } + if (exception != null) + throw new ExecutionException(exception); + return null; + } + + @Override + public synchronized Void get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + long started = System.currentTimeMillis(); + long limit = started + unit.toMillis(timeout); + while (!completed) { + long leftMs = limit - System.currentTimeMillis(); + if (leftMs < 0) + throw new TimeoutException("KafkaOffsetBackingStore Future timed out."); + this.wait(leftMs); + } + if (exception != null) + throw new ExecutionException(exception); + return null; + } + } + + private class WorkThread extends Thread { + @Override + public void run() { + try { + while (true) { + int numCallbacks; + synchronized (KafkaOffsetBackingStore.this) { + if (stopRequested) + break; + numCallbacks = readLogEndOffsetCallbacks.size(); + } + + if (numCallbacks > 0) { + try { + readToLogEnd(); + } catch (ConsumerWakeupException e) { + // Either received another get() call and need to retry reading to end of log or stop() was + // called. Both are handled by restarting this loop. + continue; + } + } + + synchronized (KafkaOffsetBackingStore.this) { + for (int i = 0; i < numCallbacks; i++) { + Callback cb = readLogEndOffsetCallbacks.poll(); + cb.onCompletion(null, null); + } + } + + try { + poll(Integer.MAX_VALUE); + } catch (ConsumerWakeupException e) { + // See previous comment, both possible causes of this wakeup are handled by starting this loop again + continue; + } + } + } catch (Throwable t) { + log.error("Unexpected exception in KafkaOffsetBackingStore's work thread", t); + } + } + } +} diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/MemoryOffsetBackingStore.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/MemoryOffsetBackingStore.java index 6ffba58293ab5..11a1b891ffd0e 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/MemoryOffsetBackingStore.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/MemoryOffsetBackingStore.java @@ -38,7 +38,7 @@ public class MemoryOffsetBackingStore implements OffsetBackingStore { private static final Logger log = LoggerFactory.getLogger(MemoryOffsetBackingStore.class); - protected HashMap> data = new HashMap<>(); + protected Map data = new HashMap<>(); protected ExecutorService executor = Executors.newSingleThreadExecutor(); public MemoryOffsetBackingStore() { @@ -60,18 +60,15 @@ public synchronized void stop() { @Override public Future> get( - final String namespace, final Collection keys, + final Collection keys, final Callback> callback) { return executor.submit(new Callable>() { @Override public Map call() throws Exception { Map result = new HashMap<>(); synchronized (MemoryOffsetBackingStore.this) { - Map namespaceData = data.get(namespace); - if (namespaceData == null) - return result; for (ByteBuffer key : keys) { - result.put(key, namespaceData.get(key)); + result.put(key, data.get(key)); } } if (callback != null) @@ -83,19 +80,14 @@ public Map call() throws Exception { } @Override - public Future set(final String namespace, final Map values, + public Future set(final Map values, final Callback callback) { return executor.submit(new Callable() { @Override public Void call() throws Exception { synchronized (MemoryOffsetBackingStore.this) { - Map namespaceData = data.get(namespace); - if (namespaceData == null) { - namespaceData = new HashMap<>(); - data.put(namespace, namespaceData); - } for (Map.Entry entry : values.entrySet()) { - namespaceData.put(entry.getKey(), entry.getValue()); + data.put(entry.getKey(), entry.getValue()); } save(); } diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetBackingStore.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetBackingStore.java index e8cb2aeb83742..239d9a8d73b76 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetBackingStore.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetBackingStore.java @@ -34,8 +34,8 @@ *

*

* Since OffsetBackingStore is a shared resource that may be used by many OffsetStorage instances - * that are associated with individual tasks, all operations include a namespace which should be - * used to isolate different key spaces. + * that are associated with individual tasks, the caller must be sure keys include information about the + * connector so that the shared namespace does not result in conflicting keys. *

*/ public interface OffsetBackingStore extends Configurable { @@ -53,22 +53,20 @@ public interface OffsetBackingStore extends Configurable { /** * Get the values for the specified keys - * @param namespace prefix for the keys in this request * @param keys list of keys to look up * @param callback callback to invoke on completion * @return future for the resulting map from key to value */ public Future> get( - String namespace, Collection keys, + Collection keys, Callback> callback); /** * Set the specified keys and values. - * @param namespace prefix for the keys in this request * @param values map from key to value * @param callback callback to invoke on completion * @return void future for the operation */ - public Future set(String namespace, Map values, + public Future set(Map values, Callback callback); } diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReaderImpl.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReaderImpl.java index 7521955abc077..dbb3d0d087304 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReaderImpl.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageReaderImpl.java @@ -62,7 +62,7 @@ public Map, Map> offsets(Collection Map, Map> offsets(Collection serialized value from backing store Map raw; try { - raw = backingStore.get(namespace, serializedToOriginal.keySet(), null).get(); + raw = backingStore.get(serializedToOriginal.keySet(), null).get(); } catch (Exception e) { log.error("Failed to fetch offsets from namespace {}: ", namespace, e); throw new CopycatException("Failed to fetch offsets.", e); diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageWriter.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageWriter.java index be8c7187308d7..59c12a7c9ade3 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageWriter.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/storage/OffsetStorageWriter.java @@ -139,7 +139,8 @@ public Future doFlush(final Callback callback) { // for that data. The only enforcement of the format is here. OffsetUtils.validateFormat(entry.getKey()); OffsetUtils.validateFormat(entry.getValue()); - byte[] key = keyConverter.fromCopycatData(namespace, null, entry.getKey()); + // When serializing the key, we add in the namespace information so the key is [namespace, real key] + byte[] key = keyConverter.fromCopycatData(namespace, null, Arrays.asList(namespace, entry.getKey())); ByteBuffer keyBuffer = (key != null) ? ByteBuffer.wrap(key) : null; byte[] value = valueConverter.fromCopycatData(namespace, null, entry.getValue()); ByteBuffer valueBuffer = (value != null) ? ByteBuffer.wrap(value) : null; @@ -158,7 +159,7 @@ public Future doFlush(final Callback callback) { // And submit the data log.debug("Submitting {} entries to backing store", offsetsSerialized.size()); - return backingStore.set(namespace, offsetsSerialized, new Callback() { + return backingStore.set(offsetsSerialized, new Callback() { @Override public void onCompletion(Throwable error, Void result) { boolean isCurrent = handleFinishWrite(flushId, error, result); diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/ConvertingFutureCallback.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/ConvertingFutureCallback.java new file mode 100644 index 0000000000000..6bf38857933e5 --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/ConvertingFutureCallback.java @@ -0,0 +1,84 @@ +/** + * 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.util; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +public abstract class ConvertingFutureCallback implements Callback, Future { + + private Callback underlying; + private CountDownLatch finishedLatch; + private T result = null; + private Throwable exception = null; + + public ConvertingFutureCallback(Callback underlying) { + this.underlying = underlying; + this.finishedLatch = new CountDownLatch(1); + } + + public abstract T convert(U result); + + @Override + public void onCompletion(Throwable error, U result) { + this.exception = error; + this.result = convert(result); + if (underlying != null) + underlying.onCompletion(error, this.result); + finishedLatch.countDown(); + } + + @Override + public boolean cancel(boolean b) { + return false; + } + + @Override + public boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return finishedLatch.getCount() == 0; + } + + @Override + public T get() throws InterruptedException, ExecutionException { + finishedLatch.await(); + return result(); + } + + @Override + public T get(long l, TimeUnit timeUnit) + throws InterruptedException, ExecutionException, TimeoutException { + finishedLatch.await(l, timeUnit); + return result(); + } + + private T result() throws ExecutionException { + if (exception != null) { + throw new ExecutionException(exception); + } + return result; + } +} + diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/FutureCallback.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/FutureCallback.java index db9f2c457621b..61e04b6d53441 100644 --- a/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/FutureCallback.java +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/util/FutureCallback.java @@ -17,60 +17,14 @@ package org.apache.kafka.copycat.util; -import java.util.concurrent.*; - -public class FutureCallback implements Callback, Future { - - private Callback underlying; - private CountDownLatch finishedLatch; - private T result = null; - private Throwable exception = null; +public class FutureCallback extends ConvertingFutureCallback { public FutureCallback(Callback underlying) { - this.underlying = underlying; - this.finishedLatch = new CountDownLatch(1); - } - - @Override - public void onCompletion(Throwable error, T result) { - underlying.onCompletion(error, result); - this.exception = error; - this.result = result; - finishedLatch.countDown(); + super(underlying); } @Override - public boolean cancel(boolean b) { - return false; - } - - @Override - public boolean isCancelled() { - return false; - } - - @Override - public boolean isDone() { - return finishedLatch.getCount() == 0; - } - - @Override - public T get() throws InterruptedException, ExecutionException { - finishedLatch.await(); - return result(); - } - - @Override - public T get(long l, TimeUnit timeUnit) - throws InterruptedException, ExecutionException, TimeoutException { - finishedLatch.await(l, timeUnit); - return result(); - } - - private T result() throws ExecutionException { - if (exception != null) { - throw new ExecutionException(exception); - } + public T convert(T result) { return result; } } 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 index 701e23023f2aa..e75d2f90d1ad4 100644 --- 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 @@ -37,6 +37,7 @@ import org.powermock.modules.junit4.PowerMockRunner; import java.util.List; +import java.util.Map; import java.util.Properties; @RunWith(PowerMockRunner.class) @@ -45,6 +46,7 @@ public class WorkerTest extends ThreadedTest { private ConnectorTaskId taskId = new ConnectorTaskId("job", 0); + private WorkerConfig config; private Worker worker; private OffsetBackingStore offsetBackingStore = PowerMock.createMock(OffsetBackingStore.class); @@ -59,13 +61,16 @@ public void setup() { workerProps.setProperty("offset.value.converter", "org.apache.kafka.copycat.json.JsonConverter"); workerProps.setProperty("offset.key.converter.schemas.enable", "false"); workerProps.setProperty("offset.value.converter.schemas.enable", "false"); - WorkerConfig config = new WorkerConfig(workerProps); - worker = new Worker(new MockTime(), config, offsetBackingStore); - worker.start(); + config = new WorkerConfig(workerProps); } @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 @@ -96,8 +101,13 @@ public void testAddRemoveTask() throws Exception { workerTask.close(); EasyMock.expectLastCall(); + offsetBackingStore.stop(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); + worker = new Worker(new MockTime(), config, offsetBackingStore); + worker.start(); worker.addTask(taskId, TestSourceTask.class.getName(), origProps); worker.stopTask(taskId); // Nothing should be left, so this should effectively be a nop @@ -108,11 +118,26 @@ public void testAddRemoveTask() throws Exception { @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(taskId); } @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); @@ -142,8 +167,13 @@ public void testCleanupTasksOnStop() throws Exception { workerTask.close(); EasyMock.expectLastCall(); + offsetBackingStore.stop(); + EasyMock.expectLastCall(); + PowerMock.replayAll(); + worker = new Worker(new MockTime(), config, offsetBackingStore); + worker.start(); worker.addTask(taskId, TestSourceTask.class.getName(), origProps); worker.stop(); diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/FileOffsetBackingStoreTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/FileOffsetBackingStoreTest.java index bbcbdc9c6064a..2976c0a7398df 100644 --- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/FileOffsetBackingStoreTest.java +++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/FileOffsetBackingStoreTest.java @@ -67,9 +67,9 @@ public void testGetSet() throws Exception { Callback> getCallback = expectSuccessfulGetCallback(); PowerMock.replayAll(); - store.set("namespace", firstSet, setCallback).get(); + store.set(firstSet, setCallback).get(); - Map values = store.get("namespace", Arrays.asList(buffer("key"), buffer("bad")), getCallback).get(); + Map values = store.get(Arrays.asList(buffer("key"), buffer("bad")), getCallback).get(); assertEquals(buffer("value"), values.get(buffer("key"))); assertEquals(null, values.get(buffer("bad"))); @@ -82,14 +82,14 @@ public void testSaveRestore() throws Exception { Callback> getCallback = expectSuccessfulGetCallback(); PowerMock.replayAll(); - store.set("namespace", firstSet, setCallback).get(); + store.set(firstSet, setCallback).get(); store.stop(); // Restore into a new store to ensure correct reload from scratch FileOffsetBackingStore restore = new FileOffsetBackingStore(); restore.configure(props); restore.start(); - Map values = restore.get("namespace", Arrays.asList(buffer("key")), getCallback).get(); + Map values = restore.get(Arrays.asList(buffer("key")), getCallback).get(); assertEquals(buffer("value"), values.get(buffer("key"))); PowerMock.verifyAll(); diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaOffsetBackingStoreTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaOffsetBackingStoreTest.java new file mode 100644 index 0000000000000..6a3eec3e48469 --- /dev/null +++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/KafkaOffsetBackingStoreTest.java @@ -0,0 +1,458 @@ +/** + * 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.storage; + +import org.apache.kafka.clients.CommonClientConfigs; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.MockConsumer; +import org.apache.kafka.clients.consumer.OffsetResetStrategy; +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.KafkaException; +import org.apache.kafka.common.Node; +import org.apache.kafka.common.PartitionInfo; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.LeaderNotAvailableException; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.util.Callback; +import org.apache.kafka.copycat.util.TestFuture; +import org.easymock.Capture; +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.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.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.apache.kafka.copycat.util.ByteArrayProducerRecordEquals.eqProducerRecord; + +@RunWith(PowerMockRunner.class) +@PrepareForTest(KafkaOffsetBackingStore.class) +@PowerMockIgnore("javax.management.*") +public class KafkaOffsetBackingStoreTest { + private static final String TOPIC = "copycat-offsets"; + private static final TopicPartition TP0 = new TopicPartition(TOPIC, 0); + private static final TopicPartition TP1 = new TopicPartition(TOPIC, 1); + 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 Set CONSUMER_ASSIGNMENT = new HashSet<>(Arrays.asList(TP0, TP1)); + private static final Map FIRST_SET = new HashMap<>(); + static { + FIRST_SET.put(buffer("key"), buffer("value")); + FIRST_SET.put(null, null); + } + + + private static final Node LEADER = new Node(1, "broker1", 9092); + private static final Node REPLICA = new Node(1, "broker2", 9093); + + private static final PartitionInfo TPINFO0 = new PartitionInfo(TOPIC, 0, LEADER, new Node[]{REPLICA}, new Node[]{REPLICA}); + private static final PartitionInfo TPINFO1 = new PartitionInfo(TOPIC, 1, LEADER, new Node[]{REPLICA}, new Node[]{REPLICA}); + + private static final ByteBuffer TP0_KEY = buffer("TP0KEY"); + private static final ByteBuffer TP1_KEY = buffer("TP1KEY"); + private static final ByteBuffer TP0_VALUE = buffer("VAL0"); + private static final ByteBuffer TP1_VALUE = buffer("VAL1"); + private static final ByteBuffer TP0_VALUE_NEW = buffer("VAL0_NEW"); + private static final ByteBuffer TP1_VALUE_NEW = buffer("VAL1_NEW"); + + private KafkaOffsetBackingStore store; + + @Mock private KafkaProducer producer; + private MockConsumer consumer; + + @Before + public void setUp() throws Exception { + store = PowerMock.createPartialMockAndInvokeDefaultConstructor(KafkaOffsetBackingStore.class, new String[]{"createConsumer", "createProducer"}); + consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); + consumer.updatePartitions(TOPIC, Arrays.asList(TPINFO0, TPINFO1)); + Map beginningOffsets = new HashMap<>(); + beginningOffsets.put(TP0, 0L); + beginningOffsets.put(TP1, 0L); + consumer.updateBeginningOffsets(beginningOffsets); + } + + @Test(expected = CopycatException.class) + public void testMissingTopic() { + store = new KafkaOffsetBackingStore(); + store.configure(Collections.emptyMap()); + } + + @Test + public void testStartStop() throws Exception { + expectStart(); + expectStop(); + + PowerMock.replayAll(); + + store.configure(DEFAULT_PROPS); + + Map endOffsets = new HashMap<>(); + endOffsets.put(TP0, 0L); + endOffsets.put(TP1, 0L); + consumer.updateEndOffsets(endOffsets); + store.start(); + assertEquals(CONSUMER_ASSIGNMENT, consumer.assignment()); + + store.stop(); + + assertFalse(Whitebox.getInternalState(store, "thread").isAlive()); + assertTrue(consumer.closed()); + PowerMock.verifyAll(); + } + + @Test + public void testReloadOnStart() throws Exception { + expectStart(); + expectStop(); + + PowerMock.replayAll(); + + store.configure(DEFAULT_PROPS); + + Map endOffsets = new HashMap<>(); + endOffsets.put(TP0, 1L); + endOffsets.put(TP1, 1L); + consumer.updateEndOffsets(endOffsets); + Thread startConsumerOpsThread = new Thread("start-consumer-ops-thread") { + @Override + public void run() { + // Needs to seek to end to find end offsets + consumer.waitForPoll(10000); + + // Should keep polling until it reaches current log end offset for all partitions + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 0, TP0_KEY.array(), TP0_VALUE.array())); + } + }, 10000); + + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 1, 0, TP1_KEY.array(), TP1_VALUE.array())); + } + }, 10000); + } + }; + startConsumerOpsThread.start(); + store.start(); + startConsumerOpsThread.join(10000); + assertFalse(startConsumerOpsThread.isAlive()); + assertEquals(CONSUMER_ASSIGNMENT, consumer.assignment()); + HashMap data = Whitebox.getInternalState(store, "data"); + assertEquals(TP0_VALUE, data.get(TP0_KEY)); + assertEquals(TP1_VALUE, data.get(TP1_KEY)); + + store.stop(); + + assertFalse(Whitebox.getInternalState(store, "thread").isAlive()); + assertTrue(consumer.closed()); + PowerMock.verifyAll(); + } + + @Test + public void testGetSet() throws Exception { + expectStart(); + TestFuture tp0Future = new TestFuture<>(); + ProducerRecord tp0Record = new ProducerRecord<>(TOPIC, TP0_KEY.array(), TP0_VALUE.array()); + Capture callback1 = EasyMock.newCapture(); + EasyMock.expect(producer.send(eqProducerRecord(tp0Record), EasyMock.capture(callback1))).andReturn(tp0Future); + TestFuture tp1Future = new TestFuture<>(); + ProducerRecord tp1Record = new ProducerRecord<>(TOPIC, TP1_KEY.array(), TP1_VALUE.array()); + Capture callback2 = EasyMock.newCapture(); + EasyMock.expect(producer.send(eqProducerRecord(tp1Record), EasyMock.capture(callback2))).andReturn(tp1Future); + + expectStop(); + + PowerMock.replayAll(); + + store.configure(DEFAULT_PROPS); + + Map endOffsets = new HashMap<>(); + endOffsets.put(TP0, 0L); + endOffsets.put(TP1, 0L); + consumer.updateEndOffsets(endOffsets); + Thread startConsumerOpsThread = new Thread("start-consumer-ops-thread") { + @Override + public void run() { + // Should keep polling until it has partition info + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.seek(TP0, 0); + consumer.seek(TP1, 0); + } + }, 10000); + } + }; + startConsumerOpsThread.start(); + store.start(); + startConsumerOpsThread.join(10000); + assertFalse(startConsumerOpsThread.isAlive()); + assertEquals(CONSUMER_ASSIGNMENT, consumer.assignment()); + + Map toSet = new HashMap<>(); + toSet.put(TP0_KEY, TP0_VALUE); + toSet.put(TP1_KEY, TP1_VALUE); + final AtomicBoolean invoked = new AtomicBoolean(false); + Future setFuture = store.set(toSet, new Callback() { + @Override + public void onCompletion(Throwable error, Void result) { + invoked.set(true); + } + }); + assertFalse(setFuture.isDone()); + tp1Future.resolve((RecordMetadata) null); // Output not used, so safe to not return a real value for testing + assertFalse(setFuture.isDone()); + tp0Future.resolve((RecordMetadata) null); + // Out of order callbacks shouldn't matter, should still require all to be invoked before invoking the callback + // for the store's set callback + callback2.getValue().onCompletion(null, null); + assertFalse(invoked.get()); + callback1.getValue().onCompletion(null, null); + setFuture.get(10000, TimeUnit.MILLISECONDS); + assertTrue(invoked.get()); + + // Getting data should continue to return old data... + final AtomicBoolean getInvokedAndPassed = new AtomicBoolean(false); + store.get(Arrays.asList(TP0_KEY, TP1_KEY), new Callback>() { + @Override + public void onCompletion(Throwable error, Map result) { + // Since we didn't read them yet, these will be null + assertEquals(null, result.get(TP0_KEY)); + assertEquals(null, result.get(TP1_KEY)); + getInvokedAndPassed.set(true); + } + }).get(10000, TimeUnit.MILLISECONDS); + assertTrue(getInvokedAndPassed.get()); + + // Until the consumer gets the new data + Thread readNewDataThread = new Thread("read-new-data-thread") { + @Override + public void run() { + // Should keep polling until it reaches current log end offset for all partitions + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 0, TP0_KEY.array(), TP0_VALUE_NEW.array())); + } + }, 10000); + + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 1, 0, TP1_KEY.array(), TP1_VALUE_NEW.array())); + } + }, 10000); + } + }; + readNewDataThread.start(); + readNewDataThread.join(10000); + assertFalse(readNewDataThread.isAlive()); + + // And now the new data should be returned + final AtomicBoolean finalGetInvokedAndPassed = new AtomicBoolean(false); + store.get(Arrays.asList(TP0_KEY, TP1_KEY), new Callback>() { + @Override + public void onCompletion(Throwable error, Map result) { + assertEquals(TP0_VALUE_NEW, result.get(TP0_KEY)); + assertEquals(TP1_VALUE_NEW, result.get(TP1_KEY)); + finalGetInvokedAndPassed.set(true); + } + }).get(10000, TimeUnit.MILLISECONDS); + assertTrue(finalGetInvokedAndPassed.get()); + + store.stop(); + + assertFalse(Whitebox.getInternalState(store, "thread").isAlive()); + assertTrue(consumer.closed()); + PowerMock.verifyAll(); + } + + @Test + public void testConsumerError() throws Exception { + expectStart(); + expectStop(); + + PowerMock.replayAll(); + + store.configure(DEFAULT_PROPS); + + Map endOffsets = new HashMap<>(); + endOffsets.put(TP0, 1L); + endOffsets.put(TP1, 1L); + consumer.updateEndOffsets(endOffsets); + Thread startConsumerOpsThread = new Thread("start-consumer-ops-thread") { + @Override + public void run() { + // Trigger exception + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.setException(Errors.CONSUMER_COORDINATOR_NOT_AVAILABLE.exception()); + } + }, 10000); + + // Should keep polling until it reaches current log end offset for all partitions + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.addRecord(new ConsumerRecord<>(TOPIC, 0, 0, TP0_KEY.array(), TP0_VALUE_NEW.array())); + consumer.addRecord(new ConsumerRecord<>(TOPIC, 1, 0, TP0_KEY.array(), TP0_VALUE_NEW.array())); + } + }, 10000); + } + }; + startConsumerOpsThread.start(); + store.start(); + startConsumerOpsThread.join(10000); + assertFalse(startConsumerOpsThread.isAlive()); + assertEquals(CONSUMER_ASSIGNMENT, consumer.assignment()); + + store.stop(); + + assertFalse(Whitebox.getInternalState(store, "thread").isAlive()); + assertTrue(consumer.closed()); + PowerMock.verifyAll(); + } + + @Test + public void testProducerError() throws Exception { + expectStart(); + TestFuture tp0Future = new TestFuture<>(); + ProducerRecord tp0Record = new ProducerRecord<>(TOPIC, TP0_KEY.array(), TP0_VALUE.array()); + Capture callback1 = EasyMock.newCapture(); + EasyMock.expect(producer.send(eqProducerRecord(tp0Record), EasyMock.capture(callback1))).andReturn(tp0Future); + TestFuture tp1Future = new TestFuture<>(); + ProducerRecord tp1Record = new ProducerRecord<>(TOPIC, TP1_KEY.array(), TP1_VALUE.array()); + Capture callback2 = EasyMock.newCapture(); + EasyMock.expect(producer.send(eqProducerRecord(tp1Record), EasyMock.capture(callback2))).andReturn(tp1Future); + + expectStop(); + + PowerMock.replayAll(); + + store.configure(DEFAULT_PROPS); + + Map endOffsets = new HashMap<>(); + endOffsets.put(TP0, 0L); + endOffsets.put(TP1, 0L); + consumer.updateEndOffsets(endOffsets); + Thread startConsumerOpsThread = new Thread("start-consumer-ops-thread") { + @Override + public void run() { + // Should keep polling until it has partition info + consumer.waitForPollThen(new Runnable() { + @Override + public void run() { + consumer.seek(TP0, 0); + consumer.seek(TP1, 0); + } + }, 10000); + } + }; + startConsumerOpsThread.start(); + store.start(); + startConsumerOpsThread.join(10000); + assertFalse(startConsumerOpsThread.isAlive()); + assertEquals(CONSUMER_ASSIGNMENT, consumer.assignment()); + + Map toSet = new HashMap<>(); + toSet.put(TP0_KEY, TP0_VALUE); + toSet.put(TP1_KEY, TP1_VALUE); + final AtomicReference setException = new AtomicReference<>(); + Future setFuture = store.set(toSet, new Callback() { + @Override + public void onCompletion(Throwable error, Void result) { + assertNull(setException.get()); // Should only be invoked once + setException.set(error); + } + }); + assertFalse(setFuture.isDone()); + KafkaException exc = new LeaderNotAvailableException("Error"); + tp1Future.resolve(exc); + callback2.getValue().onCompletion(null, exc); + // One failure should resolve the future immediately + try { + setFuture.get(10000, TimeUnit.MILLISECONDS); + fail("Should have see ExecutionException"); + } catch (ExecutionException e) { + // expected + } + assertNotNull(setException.get()); + + // Callbacks can continue to arrive + tp0Future.resolve((RecordMetadata) null); + callback1.getValue().onCompletion(null, null); + + store.stop(); + + assertFalse(Whitebox.getInternalState(store, "thread").isAlive()); + assertTrue(consumer.closed()); + PowerMock.verifyAll(); + } + + + private void expectStart() throws Exception { + PowerMock.expectPrivate(store, "createProducer") + .andReturn(producer); + PowerMock.expectPrivate(store, "createConsumer") + .andReturn(consumer); + } + + private void expectStop() { + producer.close(); + PowerMock.expectLastCall(); + // MockConsumer close is checked after test. + } + + private static ByteBuffer buffer(String v) { + return ByteBuffer.wrap(v.getBytes()); + } + +} diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java index 956d06447289a..e33ecd0d99476 100644 --- a/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java +++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/storage/OffsetStorageWriterTest.java @@ -31,7 +31,9 @@ import org.powermock.modules.junit4.PowerMockRunner; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.concurrent.*; @@ -43,6 +45,7 @@ public class OffsetStorageWriterTest { private static final String NAMESPACE = "namespace"; // Copycat format - any types should be accepted here private static final Map OFFSET_KEY = Collections.singletonMap("key", "key"); + private static final List OFFSET_KEY_WRAPPED = Arrays.asList(NAMESPACE, OFFSET_KEY); private static final Map OFFSET_VALUE = Collections.singletonMap("key", 12); // Serialized @@ -195,12 +198,11 @@ private void expectStore(final Callback callback, final boolean fail) { private void expectStore(final Callback callback, final boolean fail, final CountDownLatch waitForCompletion) { - EasyMock.expect(keyConverter.fromCopycatData(NAMESPACE, null, OFFSET_KEY)).andReturn(OFFSET_KEY_SERIALIZED); + EasyMock.expect(keyConverter.fromCopycatData(NAMESPACE, null, OFFSET_KEY_WRAPPED)).andReturn(OFFSET_KEY_SERIALIZED); EasyMock.expect(valueConverter.fromCopycatData(NAMESPACE, null, OFFSET_VALUE)).andReturn(OFFSET_VALUE_SERIALIZED); final Capture> storeCallback = Capture.newInstance(); - EasyMock.expect(store.set(EasyMock.eq(NAMESPACE), EasyMock.eq(OFFSETS_SERIALIZED), - EasyMock.capture(storeCallback))) + EasyMock.expect(store.set(EasyMock.eq(OFFSETS_SERIALIZED), EasyMock.capture(storeCallback))) .andAnswer(new IAnswer>() { @Override public Future answer() throws Throwable { diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/ByteArrayProducerRecordEquals.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/ByteArrayProducerRecordEquals.java new file mode 100644 index 0000000000000..929ea859bb14b --- /dev/null +++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/ByteArrayProducerRecordEquals.java @@ -0,0 +1,53 @@ +/** + * 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.util; + +import org.apache.kafka.clients.producer.ProducerRecord; +import org.easymock.EasyMock; +import org.easymock.IArgumentMatcher; + +import java.util.Arrays; + +public class ByteArrayProducerRecordEquals implements IArgumentMatcher { + private ProducerRecord record; + + public static ProducerRecord eqProducerRecord(ProducerRecord in) { + EasyMock.reportMatcher(new ByteArrayProducerRecordEquals(in)); + return null; + } + + public ByteArrayProducerRecordEquals(ProducerRecord record) { + this.record = record; + } + + @Override + public boolean matches(Object argument) { + if (!(argument instanceof ProducerRecord)) + return false; + ProducerRecord other = (ProducerRecord) argument; + return record.topic().equals(other.topic()) && + record.partition() != null ? record.partition().equals(other.partition()) : other.partition() == null && + record.key() != null ? Arrays.equals(record.key(), other.key()) : other.key() == null && + record.value() != null ? Arrays.equals(record.value(), other.value()) : other.value() == null; + } + + @Override + public void appendTo(StringBuffer buffer) { + buffer.append(record.toString()); + } +} diff --git a/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java b/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java new file mode 100644 index 0000000000000..8143c44a6148f --- /dev/null +++ b/copycat/runtime/src/test/java/org/apache/kafka/copycat/util/TestFuture.java @@ -0,0 +1,81 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ + +package org.apache.kafka.copycat.util; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +public class TestFuture implements Future { + private volatile boolean resolved; + private T result; + private Throwable exception; + + public TestFuture() { + resolved = false; + } + + public void resolve(T val) { + this.result = val; + resolved = true; + + } + + public void resolve(Throwable t) { + exception = t; + resolved = true; + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + + @Override + public boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return resolved; + } + + @Override + public T get() throws InterruptedException, ExecutionException { + while (true) { + try { + return get(Integer.MAX_VALUE, TimeUnit.DAYS); + } catch (TimeoutException e) { + // ignore + } + } + } + + @Override + public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + while (!resolved) { + this.wait(TimeUnit.MILLISECONDS.convert(timeout, unit)); + } + + if (exception != null) + throw new ExecutionException(exception); + return result; + } +} diff --git a/copycat/runtime/src/test/resources/log4j.properties b/copycat/runtime/src/test/resources/log4j.properties new file mode 100644 index 0000000000000..d5e90fe788f76 --- /dev/null +++ b/copycat/runtime/src/test/resources/log4j.properties @@ -0,0 +1,23 @@ +## +# 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. +## +log4j.rootLogger=OFF, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n + +log4j.logger.org.apache.kafka=ERROR diff --git a/tests/kafkatest/services/copycat.py b/tests/kafkatest/services/copycat.py index 7452b858fc319..ea5753bc5c500 100644 --- a/tests/kafkatest/services/copycat.py +++ b/tests/kafkatest/services/copycat.py @@ -18,8 +18,8 @@ import subprocess, signal -class CopycatStandaloneService(Service): - """Runs Copycat in standalone mode.""" +class CopycatServiceBase(Service): + """Base class for Copycat services providing some common settings and functionality""" logs = { "kafka_log": { @@ -27,16 +27,55 @@ class CopycatStandaloneService(Service): "collect_default": True}, } - def __init__(self, context, kafka, files): - super(CopycatStandaloneService, self).__init__(context, 1) + def __init__(self, context, num_nodes, kafka, files): + super(CopycatServiceBase, self).__init__(context, num_nodes) self.kafka = kafka self.files = files + def pids(self, node): + """Return process ids for Copycat processes.""" + try: + return [pid for pid in node.account.ssh_capture("cat /mnt/copycat.pid", callback=int)] + except: + return [] + + def stop_node(self, node, clean_shutdown=True): + pids = self.pids(node) + sig = signal.SIGTERM if clean_shutdown else signal.SIGKILL + + for pid in pids: + node.account.signal(pid, sig, allow_fail=False) + for pid in pids: + wait_until(lambda: not node.account.alive(pid), timeout_sec=10, err_msg="Copycat standalone process took too long to exit") + + node.account.ssh("rm -f /mnt/copycat.pid", allow_fail=False) + + def restart(self): + # We don't want to do any clean up here, just restart the process + for node in self.nodes: + self.stop_node(node) + self.start_node(node) + + def clean_node(self, node): + if len(self.pids(node)) > 0: + self.logger.warn("%s %s was still alive at cleanup time. Killing forcefully..." % + (self.__class__.__name__, node.account)) + for pid in self.pids(node): + node.account.signal(pid, signal.SIGKILL, allow_fail=False) + node.account.ssh("rm -rf /mnt/copycat.pid /mnt/copycat.log /mnt/copycat.properties /mnt/copycat-connector.properties " + " ".join(self.files), allow_fail=False) + + +class CopycatStandaloneService(CopycatServiceBase): + """Runs Copycat in standalone mode.""" + + def __init__(self, context, kafka, files): + super(CopycatStandaloneService, self).__init__(context, 1, kafka, files) + def set_configs(self, config_template, connector_config_template): """ - Set configurations for the standalone worker and the connector to run on - it. These are not provided in the constructor because at least the - standalone worker config generally needs access to ZK/Kafka services to + Set configurations for the worker and the connector to run on + it. These are not provided in the constructor because the worker + config generally needs access to ZK/Kafka services to create the configuration. """ self.config_template = config_template @@ -47,9 +86,6 @@ def set_configs(self, config_template, connector_config_template): def node(self): return self.nodes[0] - def start(self): - super(CopycatStandaloneService, self).start() - def start_node(self, node): node.account.create_file("/mnt/copycat.properties", self.config_template) node.account.create_file("/mnt/copycat-connector.properties", self.connector_config_template) @@ -60,36 +96,38 @@ def start_node(self, node): "1>> /mnt/copycat.log 2>> /mnt/copycat.log & echo $! > /mnt/copycat.pid") monitor.wait_until('Copycat started', timeout_sec=10, err_msg="Never saw message indicating Copycat finished startup") - if len(self.pids()) == 0: + if len(self.pids(node)) == 0: raise RuntimeError("No process ids recorded") - def pids(self): - """Return process ids for Copycat processes.""" - try: - return [pid for pid in self.node.account.ssh_capture("cat /mnt/copycat.pid", callback=int)] - except: - return [] - def stop_node(self, node, clean_shutdown=True): - pids = self.pids() - sig = signal.SIGTERM if clean_shutdown else signal.SIGKILL - for pid in pids: - self.node.account.signal(pid, sig, allow_fail=False) - for pid in pids: - wait_until(lambda: not self.node.account.alive(pid), timeout_sec=10, err_msg="Copycat standalone process took too long to exit") +class CopycatDistributedService(CopycatServiceBase): + """Runs Copycat in distributed mode.""" - node.account.ssh("rm -f /mnt/copycat.pid", allow_fail=False) + def __init__(self, context, num_nodes, kafka, files, offsets_topic="copycat-offsets"): + super(CopycatDistributedService, self).__init__(context, num_nodes, kafka, files) + self.offsets_topic = offsets_topic - def restart(self): - # We don't want to do any clean up here, just restart the process - self.stop_node(self.node) - self.start_node(self.node) + def set_configs(self, config_template, connector_config_template): + """ + Set configurations for the worker and the connector to run on + it. These are not provided in the constructor because the worker + config generally needs access to ZK/Kafka services to + create the configuration. + """ + self.config_template = config_template + self.connector_config_template = connector_config_template + + def start_node(self, node): + node.account.create_file("/mnt/copycat.properties", self.config_template) + node.account.create_file("/mnt/copycat-connector.properties", self.connector_config_template) + + self.logger.info("Starting Copycat standalone process") + with node.account.monitor_log("/mnt/copycat.log") as monitor: + node.account.ssh("/opt/kafka/bin/copycat-distributed.sh /mnt/copycat.properties /mnt/copycat-connector.properties " + + "1>> /mnt/copycat.log 2>> /mnt/copycat.log & echo $! > /mnt/copycat.pid") + monitor.wait_until('Copycat started', timeout_sec=10, err_msg="Never saw message indicating Copycat finished startup") + + if len(self.pids(node)) == 0: + raise RuntimeError("No process ids recorded") - def clean_node(self, node): - if len(self.pids()) > 0: - self.logger.warn("%s %s was still alive at cleanup time. Killing forcefully..." % - (self.__class__.__name__, node.account)) - for pid in self.pids(): - self.node.account.signal(pid, signal.SIGKILL, allow_fail=False) - node.account.ssh("rm -rf /mnt/copycat.pid /mnt/copycat.log /mnt/copycat.properties /mnt/copycat-connector.properties " + " ".join(self.files), allow_fail=False) diff --git a/tests/kafkatest/tests/copycat_distributed_test.py b/tests/kafkatest/tests/copycat_distributed_test.py new file mode 100644 index 0000000000000..3afcd57d86d54 --- /dev/null +++ b/tests/kafkatest/tests/copycat_distributed_test.py @@ -0,0 +1,84 @@ +# 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. + +from kafkatest.tests.kafka_test import KafkaTest +from kafkatest.services.copycat import CopycatDistributedService +from ducktape.utils.util import wait_until +import hashlib, subprocess, json, itertools + +class CopycatDistributedFileTest(KafkaTest): + """ + Simple test of Copycat in distributed mode, producing data from files on on Copycat cluster and consuming it on + another, validating the total output is identical to the input. + """ + + INPUT_FILE = "/mnt/copycat.input" + OUTPUT_FILE = "/mnt/copycat.output" + + TOPIC = "test" + OFFSETS_TOPIC = "copycat-offsets" + + FIRST_INPUT_LISTS = [["foo", "bar", "baz"], ["foo2", "bar2", "baz2"]] + FIRST_INPUTS = ["\n".join(input_list) + "\n" for input_list in FIRST_INPUT_LISTS] + SECOND_INPUT_LISTS = [["razz", "ma", "tazz"], ["razz2", "ma2", "tazz2"]] + SECOND_INPUTS = ["\n".join(input_list) + "\n" for input_list in SECOND_INPUT_LISTS] + + SCHEMA = { "type": "string", "optional": False } + + def __init__(self, test_context): + super(CopycatDistributedFileTest, self).__init__(test_context, num_zk=1, num_brokers=1, topics={ + 'test' : { 'partitions': 1, 'replication-factor': 1 } + }) + + self.source = CopycatDistributedService(test_context, 2, self.kafka, [self.INPUT_FILE]) + self.sink = CopycatDistributedService(test_context, 2, self.kafka, [self.OUTPUT_FILE]) + + def test_file_source_and_sink(self, converter="org.apache.kafka.copycat.json.JsonConverter", schemas=True): + assert converter != None, "converter type must be set" + # Template parameters + self.key_converter = converter + self.value_converter = converter + self.schemas = schemas + + # These need to be set + self.source.set_configs(self.render("copycat-distributed.properties"), self.render("copycat-file-source.properties")) + self.sink.set_configs(self.render("copycat-distributed.properties"), self.render("copycat-file-sink.properties")) + + self.source.start() + self.sink.start() + + # Generating data on the source node should generate new records and create new output on the sink node + for node, input in zip(self.source.nodes, self.FIRST_INPUTS): + node.account.ssh("echo -e -n " + repr(input) + " >> " + self.INPUT_FILE) + wait_until(lambda: self.validate_output(self.FIRST_INPUT_LISTS), timeout_sec=60, err_msg="Data added to input file was not seen in the output file in a reasonable amount of time.") + + # Restarting both should result in them picking up where they left off, + # only processing new data. + self.source.restart() + self.sink.restart() + + for node, input in zip(self.source.nodes, self.SECOND_INPUTS): + node.account.ssh("echo -e -n " + repr(input) + " >> " + self.INPUT_FILE) + wait_until(lambda: self.validate_output(self.FIRST_INPUT_LISTS + self.SECOND_INPUT_LISTS), timeout_sec=60, err_msg="Sink output file never converged to the same state as the input file") + + def validate_output(self, inputs): + try: + input_set = set(itertools.chain(*inputs)) + output_set = set(itertools.chain(*[ + [line.strip() for line in node.account.ssh_capture("cat " + self.OUTPUT_FILE)] for node in self.sink.nodes + ])) + return input_set == output_set + except subprocess.CalledProcessError: + return False diff --git a/tests/kafkatest/tests/copycat_test.py b/tests/kafkatest/tests/copycat_test.py index b4adf53face31..1bd8ccbae696c 100644 --- a/tests/kafkatest/tests/copycat_test.py +++ b/tests/kafkatest/tests/copycat_test.py @@ -53,7 +53,7 @@ def __init__(self, test_context): @parametrize(converter="org.apache.kafka.copycat.json.JsonConverter", schemas=True) @parametrize(converter="org.apache.kafka.copycat.json.JsonConverter", schemas=False) @parametrize(converter="org.apache.kafka.copycat.storage.StringConverter", schemas=None) - def test_file_source_and_sink(self, converter="org.apache.kafka.json.JsonConverter", schemas=True): + def test_file_source_and_sink(self, converter="org.apache.kafka.copycat.json.JsonConverter", schemas=True): assert converter != None, "converter type must be set" # Template parameters self.key_converter = converter diff --git a/tests/kafkatest/tests/templates/copycat-distributed.properties b/tests/kafkatest/tests/templates/copycat-distributed.properties new file mode 100644 index 0000000000000..da47423f56201 --- /dev/null +++ b/tests/kafkatest/tests/templates/copycat-distributed.properties @@ -0,0 +1,32 @@ +# 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. + +bootstrap.servers={{ kafka.bootstrap_servers() }} + +key.converter={{ key_converter|default("org.apache.kafka.copycat.json.JsonConverter") }} +value.converter={{ value_converter|default("org.apache.kafka.copycat.json.JsonConverter") }} +{% if key_converter is not defined or key_converter.endswith("JsonConverter") %} +key.converter.schemas.enable={{ schemas|default(True)|string|lower }} +{% endif %} +{% if value_converter is not defined or value_converter.endswith("JsonConverter") %} +value.converter.schemas.enable={{ schemas|default(True)|string|lower }} +{% endif %} + +offset.key.converter=org.apache.kafka.copycat.json.JsonConverter +offset.value.converter=org.apache.kafka.copycat.json.JsonConverter +offset.key.converter.schemas.enable=false +offset.value.converter.schemas.enable=false + +offset.storage.topic={{ OFFSETS_TOPIC }}