http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkConnector.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkConnector.java b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkConnector.java new file mode 100644 index 0000000..e41364e --- /dev/null +++ b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkConnector.java @@ -0,0 +1,62 @@ +/** + * 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.file; + +import org.apache.kafka.copycat.connector.Task; +import org.apache.kafka.copycat.sink.SinkConnector; + +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +/** + * Very simple connector that works with the console. This connector supports both source and + * sink modes via its 'mode' setting. + */ +public class FileStreamSinkConnector extends SinkConnector { + public static final String FILE_CONFIG = "file"; + + private String filename; + + @Override + public void start(Properties props) { + filename = props.getProperty(FILE_CONFIG); + } + + @Override + public Class<? extends Task> getTaskClass() { + return FileStreamSinkTask.class; + } + + @Override + public List<Properties> getTaskConfigs(int maxTasks) { + ArrayList<Properties> configs = new ArrayList<>(); + for (int i = 0; i < maxTasks; i++) { + Properties config = new Properties(); + if (filename != null) + config.setProperty(FILE_CONFIG, filename); + configs.add(config); + } + return configs; + } + + @Override + public void stop() { + // Nothing to do since FileStreamSinkConnector has no background monitoring. + } +}
http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkTask.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkTask.java b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkTask.java new file mode 100644 index 0000000..7e4ca7e --- /dev/null +++ b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSinkTask.java @@ -0,0 +1,79 @@ +/** + * 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.file; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.sink.SinkRecord; +import org.apache.kafka.copycat.sink.SinkTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.PrintStream; +import java.util.Collection; +import java.util.Map; +import java.util.Properties; + +/** + * FileStreamSinkTask writes records to stdout or a file. + */ +public class FileStreamSinkTask extends SinkTask { + private static final Logger log = LoggerFactory.getLogger(FileStreamSinkTask.class); + + private PrintStream outputStream; + + public FileStreamSinkTask() { + } + + // for testing + public FileStreamSinkTask(PrintStream outputStream) { + this.outputStream = outputStream; + } + + @Override + public void start(Properties props) { + String filename = props.getProperty(FileStreamSinkConnector.FILE_CONFIG); + if (filename == null) { + outputStream = System.out; + } else { + try { + outputStream = new PrintStream(new File(filename)); + } catch (FileNotFoundException e) { + throw new CopycatException("Couldn't find or create file for FileStreamSinkTask: {}", e); + } + } + } + + @Override + public void put(Collection<SinkRecord> sinkRecords) { + for (SinkRecord record : sinkRecords) { + outputStream.println(record.getValue()); + } + } + + @Override + public void flush(Map<TopicPartition, Long> offsets) { + outputStream.flush(); + } + + @Override + public void stop() { + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceConnector.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceConnector.java b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceConnector.java new file mode 100644 index 0000000..4f9d8d0 --- /dev/null +++ b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceConnector.java @@ -0,0 +1,70 @@ +/** + * 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.file; + +import org.apache.kafka.copycat.connector.Task; +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.source.SourceConnector; + +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +/** + * Very simple connector that works with the console. This connector supports both source and + * sink modes via its 'mode' setting. + */ +public class FileStreamSourceConnector extends SourceConnector { + public static final String TOPIC_CONFIG = "topic"; + public static final String FILE_CONFIG = "file"; + + private String filename; + private String topic; + + @Override + public void start(Properties props) { + filename = props.getProperty(FILE_CONFIG); + topic = props.getProperty(TOPIC_CONFIG); + if (topic == null || topic.isEmpty()) + throw new CopycatException("FileStreamSourceConnector configuration must include 'topic' setting"); + if (topic.contains(",")) + throw new CopycatException("FileStreamSourceConnector should only have a single topic when used as a source."); + } + + @Override + public Class<? extends Task> getTaskClass() { + return FileStreamSourceTask.class; + } + + @Override + public List<Properties> getTaskConfigs(int maxTasks) { + ArrayList<Properties> configs = new ArrayList<>(); + // Only one input stream makes sense. + Properties config = new Properties(); + if (filename != null) + config.setProperty(FILE_CONFIG, filename); + config.setProperty(TOPIC_CONFIG, topic); + configs.add(config); + return configs; + } + + @Override + public void stop() { + // Nothing to do since FileStreamSourceConnector has no background monitoring. + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceTask.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceTask.java b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceTask.java new file mode 100644 index 0000000..572ae1f --- /dev/null +++ b/copycat/file/src/main/java/org/apache/kafka/copycat/file/FileStreamSourceTask.java @@ -0,0 +1,176 @@ +/** + * 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.file; + +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.source.SourceRecord; +import org.apache.kafka.copycat.source.SourceTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.*; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; + +/** + * FileStreamSourceTask reads from stdin or a file. + */ +public class FileStreamSourceTask extends SourceTask { + private static final Logger log = LoggerFactory.getLogger(FileStreamSourceTask.class); + + private InputStream stream; + private BufferedReader reader = null; + private char[] buffer = new char[1024]; + private int offset = 0; + private String topic = null; + + private Long streamOffset; + + @Override + public void start(Properties props) { + String filename = props.getProperty(FileStreamSourceConnector.FILE_CONFIG); + if (filename == null) { + stream = System.in; + // Tracking offset for stdin doesn't make sense + streamOffset = null; + } else { + try { + stream = new FileInputStream(filename); + Long lastRecordedOffset = (Long) context.getOffsetStorageReader().getOffset(null); + if (lastRecordedOffset != null) { + log.debug("Found previous offset, trying to skip to file offset {}", lastRecordedOffset); + long skipLeft = lastRecordedOffset; + while (skipLeft > 0) { + try { + long skipped = stream.skip(skipLeft); + skipLeft -= skipped; + } catch (IOException e) { + log.error("Error while trying to seek to previous offset in file: ", e); + throw new CopycatException(e); + } + } + log.debug("Skipped to offset {}", lastRecordedOffset); + } + streamOffset = (lastRecordedOffset != null) ? lastRecordedOffset : 0L; + } catch (FileNotFoundException e) { + throw new CopycatException("Couldn't find file for FileStreamSourceTask: {}", e); + } + } + topic = props.getProperty(FileStreamSourceConnector.TOPIC_CONFIG); + if (topic == null) + throw new CopycatException("ConsoleSourceTask config missing topic setting"); + reader = new BufferedReader(new InputStreamReader(stream)); + } + + @Override + public List<SourceRecord> poll() throws InterruptedException { + // Unfortunately we can't just use readLine() because it blocks in an uninterruptible way. + // Instead we have to manage splitting lines ourselves, using simple backoff when no new data + // is available. + try { + final BufferedReader readerCopy; + synchronized (this) { + readerCopy = reader; + } + if (readerCopy == null) + return null; + + ArrayList<SourceRecord> records = null; + + int nread = 0; + while (readerCopy.ready()) { + nread = readerCopy.read(buffer, offset, buffer.length - offset); + + if (nread > 0) { + offset += nread; + if (offset == buffer.length) { + char[] newbuf = new char[buffer.length * 2]; + System.arraycopy(buffer, 0, newbuf, 0, buffer.length); + buffer = newbuf; + } + + String line; + do { + line = extractLine(); + if (line != null) { + if (records == null) + records = new ArrayList<>(); + records.add(new SourceRecord(null, streamOffset, topic, line)); + } + new ArrayList<SourceRecord>(); + } while (line != null); + } + } + + if (nread <= 0) + Thread.sleep(1); + + return records; + } catch (IOException e) { + // Underlying stream was killed, probably as a result of calling stop. Allow to return + // null, and driving thread will handle any shutdown if necessary. + } + return null; + } + + private String extractLine() { + int until = -1, newStart = -1; + for (int i = 0; i < offset; i++) { + if (buffer[i] == '\n') { + until = i; + newStart = i + 1; + break; + } else if (buffer[i] == '\r') { + // We need to check for \r\n, so we must skip this if we can't check the next char + if (i + 1 >= offset) + return null; + + until = i; + newStart = (buffer[i + 1] == '\n') ? i + 2 : i + 1; + break; + } + } + + if (until != -1) { + String result = new String(buffer, 0, until); + System.arraycopy(buffer, newStart, buffer, 0, buffer.length - newStart); + offset = offset - newStart; + if (streamOffset != null) + streamOffset += newStart; + return result; + } else { + return null; + } + } + + @Override + public void stop() { + log.trace("Stopping"); + synchronized (this) { + try { + stream.close(); + log.trace("Closed input stream"); + } catch (IOException e) { + log.error("Failed to close ConsoleSourceTask stream: ", e); + } + reader = null; + stream = null; + } + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkConnectorTest.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkConnectorTest.java b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkConnectorTest.java new file mode 100644 index 0000000..643fb43 --- /dev/null +++ b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkConnectorTest.java @@ -0,0 +1,85 @@ +/** + * 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.file; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.copycat.connector.ConnectorContext; +import org.apache.kafka.copycat.sink.SinkConnector; +import org.junit.Before; +import org.junit.Test; +import org.powermock.api.easymock.PowerMock; + +import java.util.Arrays; +import java.util.List; +import java.util.Properties; + +import static org.junit.Assert.assertEquals; + +public class FileStreamSinkConnectorTest { + + private static final String MULTIPLE_TOPICS = "test1,test2"; + private static final String[] MULTIPLE_TOPICS_LIST + = MULTIPLE_TOPICS.split(","); + private static final List<TopicPartition> MULTIPLE_TOPICS_PARTITIONS = Arrays.asList( + new TopicPartition("test1", 1), new TopicPartition("test2", 2) + ); + private static final String FILENAME = "/afilename"; + + private FileStreamSinkConnector connector; + private ConnectorContext ctx; + private Properties sinkProperties; + + @Before + public void setup() { + connector = new FileStreamSinkConnector(); + ctx = PowerMock.createMock(ConnectorContext.class); + connector.initialize(ctx); + + sinkProperties = new Properties(); + sinkProperties.setProperty(SinkConnector.TOPICS_CONFIG, MULTIPLE_TOPICS); + sinkProperties.setProperty(FileStreamSinkConnector.FILE_CONFIG, FILENAME); + } + + @Test + public void testSinkTasks() { + PowerMock.replayAll(); + + connector.start(sinkProperties); + List<Properties> taskConfigs = connector.getTaskConfigs(1); + assertEquals(1, taskConfigs.size()); + assertEquals(FILENAME, taskConfigs.get(0).getProperty(FileStreamSinkConnector.FILE_CONFIG)); + + taskConfigs = connector.getTaskConfigs(2); + assertEquals(2, taskConfigs.size()); + for (int i = 0; i < 2; i++) { + assertEquals(FILENAME, taskConfigs.get(0).getProperty(FileStreamSinkConnector.FILE_CONFIG)); + } + + PowerMock.verifyAll(); + } + + @Test + public void testTaskClass() { + PowerMock.replayAll(); + + connector.start(sinkProperties); + assertEquals(FileStreamSinkTask.class, connector.getTaskClass()); + + PowerMock.verifyAll(); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkTaskTest.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkTaskTest.java b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkTaskTest.java new file mode 100644 index 0000000..b4e1b0c --- /dev/null +++ b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSinkTaskTest.java @@ -0,0 +1,67 @@ +/** + * 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.file; + +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.copycat.sink.SinkRecord; +import org.junit.Before; +import org.junit.Test; + +import java.io.ByteArrayOutputStream; +import java.io.PrintStream; +import java.util.Arrays; +import java.util.HashMap; + +import static org.junit.Assert.assertEquals; + +public class FileStreamSinkTaskTest { + + private FileStreamSinkTask task; + private ByteArrayOutputStream os; + private PrintStream printStream; + + @Before + public void setup() { + os = new ByteArrayOutputStream(); + printStream = new PrintStream(os); + task = new FileStreamSinkTask(printStream); + } + + @Test + public void testPutFlush() { + HashMap<TopicPartition, Long> offsets = new HashMap<>(); + + // We do not call task.start() since it would override the output stream + + task.put(Arrays.asList( + new SinkRecord("topic1", 0, null, "line1", 1) + )); + offsets.put(new TopicPartition("topic1", 0), 1L); + task.flush(offsets); + assertEquals("line1\n", os.toString()); + + task.put(Arrays.asList( + new SinkRecord("topic1", 0, null, "line2", 2), + new SinkRecord("topic2", 0, null, "line3", 1) + )); + offsets.put(new TopicPartition("topic1", 0), 2L); + offsets.put(new TopicPartition("topic2", 0), 1L); + task.flush(offsets); + assertEquals("line1\nline2\nline3\n", os.toString()); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceConnectorTest.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceConnectorTest.java b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceConnectorTest.java new file mode 100644 index 0000000..e23055c --- /dev/null +++ b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceConnectorTest.java @@ -0,0 +1,104 @@ +/** + * 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.file; + +import org.apache.kafka.copycat.connector.ConnectorContext; +import org.apache.kafka.copycat.errors.CopycatException; +import org.junit.Before; +import org.junit.Test; +import org.powermock.api.easymock.PowerMock; + +import java.util.List; +import java.util.Properties; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; + +public class FileStreamSourceConnectorTest { + + private static final String SINGLE_TOPIC = "test"; + private static final String MULTIPLE_TOPICS = "test1,test2"; + private static final String FILENAME = "/somefilename"; + + private FileStreamSourceConnector connector; + private ConnectorContext ctx; + private Properties sourceProperties; + + @Before + public void setup() { + connector = new FileStreamSourceConnector(); + ctx = PowerMock.createMock(ConnectorContext.class); + connector.initialize(ctx); + + sourceProperties = new Properties(); + sourceProperties.setProperty(FileStreamSourceConnector.TOPIC_CONFIG, SINGLE_TOPIC); + sourceProperties.setProperty(FileStreamSourceConnector.FILE_CONFIG, FILENAME); + } + + @Test + public void testSourceTasks() { + PowerMock.replayAll(); + + connector.start(sourceProperties); + List<Properties> taskConfigs = connector.getTaskConfigs(1); + assertEquals(1, taskConfigs.size()); + assertEquals(FILENAME, + taskConfigs.get(0).getProperty(FileStreamSourceConnector.FILE_CONFIG)); + assertEquals(SINGLE_TOPIC, + taskConfigs.get(0).getProperty(FileStreamSourceConnector.TOPIC_CONFIG)); + + // Should be able to return fewer than requested # + taskConfigs = connector.getTaskConfigs(2); + assertEquals(1, taskConfigs.size()); + assertEquals(FILENAME, + taskConfigs.get(0).getProperty(FileStreamSourceConnector.FILE_CONFIG)); + assertEquals(SINGLE_TOPIC, + taskConfigs.get(0).getProperty(FileStreamSourceConnector.TOPIC_CONFIG)); + + PowerMock.verifyAll(); + } + + @Test + public void testSourceTasksStdin() { + PowerMock.replayAll(); + + sourceProperties.remove(FileStreamSourceConnector.FILE_CONFIG); + connector.start(sourceProperties); + List<Properties> taskConfigs = connector.getTaskConfigs(1); + assertEquals(1, taskConfigs.size()); + assertNull(taskConfigs.get(0).getProperty(FileStreamSourceConnector.FILE_CONFIG)); + + PowerMock.verifyAll(); + } + + @Test(expected = CopycatException.class) + public void testMultipleSourcesInvalid() { + sourceProperties.setProperty(FileStreamSourceConnector.TOPIC_CONFIG, MULTIPLE_TOPICS); + connector.start(sourceProperties); + } + + @Test + public void testTaskClass() { + PowerMock.replayAll(); + + connector.start(sourceProperties); + assertEquals(FileStreamSourceTask.class, connector.getTaskClass()); + + PowerMock.verifyAll(); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceTaskTest.java ---------------------------------------------------------------------- diff --git a/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceTaskTest.java b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceTaskTest.java new file mode 100644 index 0000000..0ec71d3 --- /dev/null +++ b/copycat/file/src/test/java/org/apache/kafka/copycat/file/FileStreamSourceTaskTest.java @@ -0,0 +1,140 @@ +/** + * 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.file; + +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.source.SourceRecord; +import org.apache.kafka.copycat.source.SourceTaskContext; +import org.apache.kafka.copycat.storage.OffsetStorageReader; +import org.easymock.EasyMock; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.powermock.api.easymock.PowerMock; + +import java.io.File; +import java.io.FileOutputStream; +import java.io.IOException; +import java.util.List; +import java.util.Properties; + +import static org.junit.Assert.assertEquals; + +public class FileStreamSourceTaskTest { + + private static final String TOPIC = "test"; + + private File tempFile; + private Properties config; + private OffsetStorageReader offsetStorageReader; + private FileStreamSourceTask task; + + private boolean verifyMocks = false; + + @Before + public void setup() throws IOException { + tempFile = File.createTempFile("file-stream-source-task-test", null); + config = new Properties(); + config.setProperty(FileStreamSourceConnector.FILE_CONFIG, tempFile.getAbsolutePath()); + config.setProperty(FileStreamSourceConnector.TOPIC_CONFIG, TOPIC); + task = new FileStreamSourceTask(); + offsetStorageReader = PowerMock.createMock(OffsetStorageReader.class); + task.initialize(new SourceTaskContext(offsetStorageReader)); + } + + @After + public void teardown() { + tempFile.delete(); + + if (verifyMocks) + PowerMock.verifyAll(); + } + + private void replay() { + PowerMock.replayAll(); + verifyMocks = true; + } + + @Test + public void testNormalLifecycle() throws InterruptedException, IOException { + expectOffsetLookupReturnNone(); + replay(); + + task.start(config); + + FileOutputStream os = new FileOutputStream(tempFile); + assertEquals(null, task.poll()); + os.write("partial line".getBytes()); + os.flush(); + assertEquals(null, task.poll()); + os.write(" finished\n".getBytes()); + os.flush(); + List<SourceRecord> records = task.poll(); + assertEquals(1, records.size()); + assertEquals(TOPIC, records.get(0).getTopic()); + assertEquals("partial line finished", records.get(0).getValue()); + assertEquals(22L, records.get(0).getSourceOffset()); + assertEquals(null, task.poll()); + + // Different line endings, and make sure the final \r doesn't result in a line until we can + // read the subsequent byte. + os.write("line1\rline2\r\nline3\nline4\n\r".getBytes()); + os.flush(); + records = task.poll(); + assertEquals(4, records.size()); + assertEquals("line1", records.get(0).getValue()); + assertEquals(28L, records.get(0).getSourceOffset()); + assertEquals("line2", records.get(1).getValue()); + assertEquals(35L, records.get(1).getSourceOffset()); + assertEquals("line3", records.get(2).getValue()); + assertEquals(41L, records.get(2).getSourceOffset()); + assertEquals("line4", records.get(3).getValue()); + assertEquals(47L, records.get(3).getSourceOffset()); + + os.write("subsequent text".getBytes()); + os.flush(); + records = task.poll(); + assertEquals(1, records.size()); + assertEquals("", records.get(0).getValue()); + assertEquals(48L, records.get(0).getSourceOffset()); + + task.stop(); + } + + @Test(expected = CopycatException.class) + public void testMissingTopic() { + expectOffsetLookupReturnNone(); + replay(); + + config.remove(FileStreamSourceConnector.TOPIC_CONFIG); + task.start(config); + } + + @Test(expected = CopycatException.class) + public void testInvalidFile() { + config.setProperty(FileStreamSourceConnector.FILE_CONFIG, "bogusfilename"); + task.start(config); + } + + + private void expectOffsetLookupReturnNone() { + EasyMock.expect( + offsetStorageReader.getOffset(EasyMock.anyObject(Object.class))) + .andReturn(null); + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java ---------------------------------------------------------------------- diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java new file mode 100644 index 0000000..36a6ca8 --- /dev/null +++ b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonConverter.java @@ -0,0 +1,265 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ + +package org.apache.kafka.copycat.json; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.kafka.copycat.data.*; +import org.apache.kafka.copycat.errors.CopycatException; +import org.apache.kafka.copycat.storage.Converter; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.*; + +/** + * Implementation of Converter that uses JSON to store schemas and objects. + */ +public class JsonConverter implements Converter<JsonNode> { + + private static final HashMap<String, JsonToCopycatTypeConverter> TO_COPYCAT_CONVERTERS + = new HashMap<>(); + + static { + TO_COPYCAT_CONVERTERS.put(JsonSchema.BOOLEAN_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + return value.booleanValue(); + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.INT_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + return value.intValue(); + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.LONG_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + return value.longValue(); + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.FLOAT_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + return value.floatValue(); + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.DOUBLE_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + return value.doubleValue(); + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.BYTES_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + try { + return value.binaryValue(); + } catch (IOException e) { + throw new CopycatException("Invalid bytes field", e); + } + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.STRING_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + return value.textValue(); + } + }); + TO_COPYCAT_CONVERTERS.put(JsonSchema.ARRAY_TYPE_NAME, new JsonToCopycatTypeConverter() { + @Override + public Object convert(JsonNode jsonSchema, JsonNode value) { + JsonNode elemSchema = jsonSchema.get(JsonSchema.ARRAY_ITEMS_FIELD_NAME); + if (elemSchema == null) + throw new CopycatException("Array schema did not specify the element type"); + ArrayList<Object> result = new ArrayList<>(); + for (JsonNode elem : value) { + result.add(convertToCopycat(elemSchema, elem)); + } + return result; + } + }); + + } + + @Override + public JsonNode fromCopycatData(Object value) { + return convertToJsonWithSchemaEnvelope(value); + } + + @Override + public Object toCopycatData(JsonNode value) { + if (!value.isObject() || value.size() != 2 || !value.has(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME) || !value.has(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME)) + throw new CopycatException("JSON value converted to Copycat must be in envelope containing schema"); + + return convertToCopycat(value.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME), value.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME)); + } + + + private static JsonNode asJsonSchema(Schema schema) { + switch (schema.getType()) { + case BOOLEAN: + return JsonSchema.BOOLEAN_SCHEMA; + case BYTES: + return JsonSchema.BYTES_SCHEMA; + case DOUBLE: + return JsonSchema.DOUBLE_SCHEMA; + case FLOAT: + return JsonSchema.FLOAT_SCHEMA; + case INT: + return JsonSchema.INT_SCHEMA; + case LONG: + return JsonSchema.LONG_SCHEMA; + case NULL: + throw new UnsupportedOperationException("null schema not supported"); + case STRING: + return JsonSchema.STRING_SCHEMA; + case UNION: { + throw new UnsupportedOperationException("union schema not supported"); + } + case ARRAY: + return JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME, JsonSchema.ARRAY_TYPE_NAME) + .set(JsonSchema.ARRAY_ITEMS_FIELD_NAME, asJsonSchema(schema.getElementType())); + case ENUM: + throw new UnsupportedOperationException("enum schema not supported"); + case MAP: + throw new UnsupportedOperationException("map schema not supported"); + default: + throw new CopycatException("Couldn't translate unsupported schema type " + schema.getType().getName() + "."); + } + } + + + private static Schema asCopycatSchema(JsonNode jsonSchema) { + if (jsonSchema.isNull()) + return null; + + JsonNode schemaTypeNode = jsonSchema.get(JsonSchema.SCHEMA_TYPE_FIELD_NAME); + if (schemaTypeNode == null || !schemaTypeNode.isTextual()) + throw new CopycatException("Schema must contain 'type' field"); + + switch (schemaTypeNode.textValue()) { + case JsonSchema.BOOLEAN_TYPE_NAME: + return SchemaBuilder.builder().booleanType(); + case JsonSchema.INT_TYPE_NAME: + return SchemaBuilder.builder().intType(); + case JsonSchema.LONG_TYPE_NAME: + return SchemaBuilder.builder().longType(); + case JsonSchema.FLOAT_TYPE_NAME: + return SchemaBuilder.builder().floatType(); + case JsonSchema.DOUBLE_TYPE_NAME: + return SchemaBuilder.builder().doubleType(); + case JsonSchema.BYTES_TYPE_NAME: + return SchemaBuilder.builder().bytesType(); + case JsonSchema.STRING_TYPE_NAME: + return SchemaBuilder.builder().stringType(); + case JsonSchema.ARRAY_TYPE_NAME: + JsonNode elemSchema = jsonSchema.get(JsonSchema.ARRAY_ITEMS_FIELD_NAME); + if (elemSchema == null) + throw new CopycatException("Array schema did not specify the element type"); + return Schema.createArray(asCopycatSchema(elemSchema)); + default: + throw new CopycatException("Unknown schema type: " + schemaTypeNode.textValue()); + } + } + + + /** + * Convert this object, in org.apache.kafka.copycat.data format, into a JSON object with an envelope object + * containing schema and payload fields. + * @param value + * @return + */ + private static JsonNode convertToJsonWithSchemaEnvelope(Object value) { + return convertToJson(value).toJsonNode(); + } + + /** + * Convert this object, in the org.apache.kafka.copycat.data format, into a JSON object, returning both the schema + * and the converted object. + */ + private static JsonSchema.Envelope convertToJson(Object value) { + if (value == null) { + return JsonSchema.nullEnvelope(); + } else if (value instanceof Boolean) { + return JsonSchema.booleanEnvelope((Boolean) value); + } else if (value instanceof Byte) { + return JsonSchema.intEnvelope((Byte) value); + } else if (value instanceof Short) { + return JsonSchema.intEnvelope((Short) value); + } else if (value instanceof Integer) { + return JsonSchema.intEnvelope((Integer) value); + } else if (value instanceof Long) { + return JsonSchema.longEnvelope((Long) value); + } else if (value instanceof Float) { + return JsonSchema.floatEnvelope((Float) value); + } else if (value instanceof Double) { + return JsonSchema.doubleEnvelope((Double) value); + } else if (value instanceof byte[]) { + return JsonSchema.bytesEnvelope((byte[]) value); + } else if (value instanceof ByteBuffer) { + return JsonSchema.bytesEnvelope(((ByteBuffer) value).array()); + } else if (value instanceof CharSequence) { + return JsonSchema.stringEnvelope(value.toString()); + } else if (value instanceof Collection) { + Collection collection = (Collection) value; + ObjectNode schema = JsonNodeFactory.instance.objectNode().put(JsonSchema.SCHEMA_TYPE_FIELD_NAME, JsonSchema.ARRAY_TYPE_NAME); + ArrayNode list = JsonNodeFactory.instance.arrayNode(); + JsonNode itemSchema = null; + for (Object elem : collection) { + JsonSchema.Envelope fieldSchemaAndValue = convertToJson(elem); + if (itemSchema == null) { + itemSchema = fieldSchemaAndValue.schema; + schema.set(JsonSchema.ARRAY_ITEMS_FIELD_NAME, itemSchema); + } else { + if (!itemSchema.equals(fieldSchemaAndValue.schema)) + throw new CopycatException("Mismatching schemas found in a list."); + } + + list.add(fieldSchemaAndValue.payload); + } + return new JsonSchema.Envelope(schema, list); + } + + throw new CopycatException("Couldn't convert " + value + " to Avro."); + } + + + private static Object convertToCopycat(JsonNode jsonSchema, JsonNode jsonValue) { + if (jsonSchema.isNull()) + return null; + + JsonNode schemaTypeNode = jsonSchema.get(JsonSchema.SCHEMA_TYPE_FIELD_NAME); + if (schemaTypeNode == null || !schemaTypeNode.isTextual()) + throw new CopycatException("Schema must contain 'type' field. Schema: " + jsonSchema.toString()); + + JsonToCopycatTypeConverter typeConverter = TO_COPYCAT_CONVERTERS.get(schemaTypeNode.textValue()); + if (typeConverter != null) + return typeConverter.convert(jsonSchema, jsonValue); + + throw new CopycatException("Unknown schema type: " + schemaTypeNode); + } + + + private interface JsonToCopycatTypeConverter { + Object convert(JsonNode schema, JsonNode value); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java ---------------------------------------------------------------------- diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.java new file mode 100644 index 0000000..29c7bac --- /dev/null +++ b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonDeserializer.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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ +package org.apache.kafka.copycat.json; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ArrayNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.serialization.Deserializer; + +import java.util.Map; + +/** + * JSON deserializer for Jackson's JsonNode tree model. Using the tree model allows it to work with arbitrarily + * structured data without having associated Java classes. This deserializer also supports Copycat schemas. + */ +public class JsonDeserializer implements Deserializer<JsonNode> { + private static final ObjectNode CATCH_ALL_OBJECT_SCHEMA = JsonNodeFactory.instance.objectNode(); + private static final ObjectNode CATCH_ALL_ARRAY_SCHEMA = JsonNodeFactory.instance.objectNode(); + private static final ArrayNode ALL_SCHEMAS_LIST = JsonNodeFactory.instance.arrayNode(); + private static final ObjectNode CATCH_ALL_SCHEMA = JsonNodeFactory.instance.objectNode(); + static { + CATCH_ALL_OBJECT_SCHEMA.put("type", "object") + .putArray("field").add(JsonNodeFactory.instance.objectNode().put("*", "all")); + + CATCH_ALL_ARRAY_SCHEMA.put("type", "array").put("items", "all"); + + ALL_SCHEMAS_LIST.add("boolean").add("int").add("long").add("float").add("double").add("bytes").add("string") + .add(CATCH_ALL_ARRAY_SCHEMA).add(CATCH_ALL_OBJECT_SCHEMA); + + CATCH_ALL_SCHEMA.put("name", "all").set("type", ALL_SCHEMAS_LIST); + } + + private ObjectMapper objectMapper = new ObjectMapper(); + + /** + * Default constructor needed by Kafka + */ + public JsonDeserializer() { + } + + @Override + public void configure(Map<String, ?> props, boolean isKey) { + } + + @Override + public JsonNode deserialize(String topic, byte[] bytes) { + JsonNode data; + try { + data = objectMapper.readTree(bytes); + } catch (Exception e) { + throw new SerializationException(e); + } + + // The deserialized data should either be an envelope object containing the schema and the payload or the schema + // was stripped during serialization and we need to fill in an all-encompassing schema. + if (!data.isObject() || data.size() != 2 || !data.has("schema") || !data.has("payload")) { + ObjectNode envelope = JsonNodeFactory.instance.objectNode(); + envelope.set("schema", CATCH_ALL_SCHEMA); + envelope.set("payload", data); + data = envelope; + } + + return data; + } + + @Override + public void close() { + + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java ---------------------------------------------------------------------- diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java new file mode 100644 index 0000000..a807e0f --- /dev/null +++ b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSchema.java @@ -0,0 +1,114 @@ +/** + * 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.json; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import com.fasterxml.jackson.databind.node.ObjectNode; + +import java.nio.ByteBuffer; + +public class JsonSchema { + + static final String ENVELOPE_SCHEMA_FIELD_NAME = "schema"; + static final String ENVELOPE_PAYLOAD_FIELD_NAME = "payload"; + static final String SCHEMA_TYPE_FIELD_NAME = "type"; + static final String SCHEMA_NAME_FIELD_NAME = "name"; + static final String ARRAY_ITEMS_FIELD_NAME = "items"; + static final String BOOLEAN_TYPE_NAME = "boolean"; + static final JsonNode BOOLEAN_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, BOOLEAN_TYPE_NAME); + static final String INT_TYPE_NAME = "int"; + static final JsonNode INT_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, INT_TYPE_NAME); + static final String LONG_TYPE_NAME = "long"; + static final JsonNode LONG_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, LONG_TYPE_NAME); + static final String FLOAT_TYPE_NAME = "float"; + static final JsonNode FLOAT_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, FLOAT_TYPE_NAME); + static final String DOUBLE_TYPE_NAME = "double"; + static final JsonNode DOUBLE_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, DOUBLE_TYPE_NAME); + static final String BYTES_TYPE_NAME = "bytes"; + static final JsonNode BYTES_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, BYTES_TYPE_NAME); + static final String STRING_TYPE_NAME = "string"; + static final JsonNode STRING_SCHEMA = JsonNodeFactory.instance.objectNode().put(SCHEMA_TYPE_FIELD_NAME, STRING_TYPE_NAME); + static final String ARRAY_TYPE_NAME = "array"; + + public static ObjectNode envelope(JsonNode schema, JsonNode payload) { + ObjectNode result = JsonNodeFactory.instance.objectNode(); + result.set(ENVELOPE_SCHEMA_FIELD_NAME, schema); + result.set(ENVELOPE_PAYLOAD_FIELD_NAME, payload); + return result; + } + + static class Envelope { + public JsonNode schema; + public JsonNode payload; + + public Envelope(JsonNode schema, JsonNode payload) { + this.schema = schema; + this.payload = payload; + } + + public ObjectNode toJsonNode() { + return envelope(schema, payload); + } + } + + + public static Envelope nullEnvelope() { + return new Envelope(null, null); + } + + public static Envelope booleanEnvelope(boolean value) { + return new Envelope(JsonSchema.BOOLEAN_SCHEMA, JsonNodeFactory.instance.booleanNode(value)); + } + + public static Envelope intEnvelope(byte value) { + return new Envelope(JsonSchema.INT_SCHEMA, JsonNodeFactory.instance.numberNode(value)); + } + + public static Envelope intEnvelope(short value) { + return new Envelope(JsonSchema.INT_SCHEMA, JsonNodeFactory.instance.numberNode(value)); + } + + public static Envelope intEnvelope(int value) { + return new Envelope(JsonSchema.INT_SCHEMA, JsonNodeFactory.instance.numberNode(value)); + } + + public static Envelope longEnvelope(long value) { + return new Envelope(JsonSchema.LONG_SCHEMA, JsonNodeFactory.instance.numberNode(value)); + } + + public static Envelope floatEnvelope(float value) { + return new Envelope(JsonSchema.FLOAT_SCHEMA, JsonNodeFactory.instance.numberNode(value)); + } + + public static Envelope doubleEnvelope(double value) { + return new Envelope(JsonSchema.DOUBLE_SCHEMA, JsonNodeFactory.instance.numberNode(value)); + } + + public static Envelope bytesEnvelope(byte[] value) { + return new Envelope(JsonSchema.BYTES_SCHEMA, JsonNodeFactory.instance.binaryNode(value)); + } + + public static Envelope bytesEnvelope(ByteBuffer value) { + return new Envelope(JsonSchema.BYTES_SCHEMA, JsonNodeFactory.instance.binaryNode(value.array())); + } + + public static Envelope stringEnvelope(CharSequence value) { + return new Envelope(JsonSchema.STRING_SCHEMA, JsonNodeFactory.instance.textNode(value.toString())); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java ---------------------------------------------------------------------- diff --git a/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java new file mode 100644 index 0000000..dcac270 --- /dev/null +++ b/copycat/json/src/main/java/org/apache/kafka/copycat/json/JsonSerializer.java @@ -0,0 +1,72 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ +package org.apache.kafka.copycat.json; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.serialization.Serializer; + +import java.util.Map; + +/** + * Serialize Jackson JsonNode tree model objects to UTF-8 JSON. Using the tree model allows handling arbitrarily + * structured data without corresponding Java classes. This serializer also supports Copycat schemas. + */ +public class JsonSerializer implements Serializer<JsonNode> { + + private static final String SCHEMAS_ENABLE_CONFIG = "schemas.enable"; + private static final boolean SCHEMAS_ENABLE_DEFAULT = true; + + private final ObjectMapper objectMapper = new ObjectMapper(); + private boolean enableSchemas = SCHEMAS_ENABLE_DEFAULT; + + /** + * Default constructor needed by Kafka + */ + public JsonSerializer() { + + } + + @Override + public void configure(Map<String, ?> config, boolean isKey) { + Object enableConfigsVal = config.get(SCHEMAS_ENABLE_CONFIG); + if (enableConfigsVal != null) + enableSchemas = enableConfigsVal.toString().equals("true"); + } + + @Override + public byte[] serialize(String topic, JsonNode data) { + // This serializer works for Copycat data that requires a schema to be included, so we expect it to have a + // specific format: { "schema": {...}, "payload": ... }. + if (!data.isObject() || data.size() != 2 || !data.has("schema") || !data.has("payload")) + throw new SerializationException("JsonSerializer requires \"schema\" and \"payload\" fields and may not contain additional fields"); + + try { + if (!enableSchemas) + data = data.get("payload"); + return objectMapper.writeValueAsBytes(data); + } catch (Exception e) { + throw new SerializationException("Error serializing JSON message", e); + } + } + + @Override + public void close() { + } + +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java ---------------------------------------------------------------------- diff --git a/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java b/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java new file mode 100644 index 0000000..1a725c9 --- /dev/null +++ b/copycat/json/src/test/java/org/apache/kafka/copycat/json/JsonConverterTest.java @@ -0,0 +1,173 @@ +/** + * 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.json; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.JsonNodeFactory; +import org.junit.Test; + +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.nio.ByteBuffer; +import java.util.Arrays; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +public class JsonConverterTest { + + ObjectMapper objectMapper = new ObjectMapper(); + JsonConverter converter = new JsonConverter(); + + @Test + public void booleanToCopycat() { + assertEquals(true, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"boolean\" }, \"payload\": true }"))); + assertEquals(false, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"boolean\" }, \"payload\": false }"))); + } + + @Test + public void intToCopycat() { + assertEquals(12, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"int\" }, \"payload\": 12 }"))); + } + + @Test + public void longToCopycat() { + assertEquals(12L, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"long\" }, \"payload\": 12 }"))); + assertEquals(4398046511104L, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"long\" }, \"payload\": 4398046511104 }"))); + } + + @Test + public void floatToCopycat() { + assertEquals(12.34f, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"float\" }, \"payload\": 12.34 }"))); + } + + @Test + public void doubleToCopycat() { + assertEquals(12.34, converter.toCopycatData(parse("{ \"schema\": { \"type\": \"double\" }, \"payload\": 12.34 }"))); + } + + + @Test + public void bytesToCopycat() throws UnsupportedEncodingException { + ByteBuffer reference = ByteBuffer.wrap("test-string".getBytes("UTF-8")); + String msg = "{ \"schema\": { \"type\": \"bytes\" }, \"payload\": \"dGVzdC1zdHJpbmc=\" }"; + ByteBuffer converted = ByteBuffer.wrap((byte[]) converter.toCopycatData(parse(msg))); + assertEquals(reference, converted); + } + + @Test + public void stringToCopycat() { + assertEquals("foo-bar-baz", converter.toCopycatData(parse("{ \"schema\": { \"type\": \"string\" }, \"payload\": \"foo-bar-baz\" }"))); + } + + @Test + public void arrayToCopycat() { + JsonNode arrayJson = parse("{ \"schema\": { \"type\": \"array\", \"items\": { \"type\" : \"int\" } }, \"payload\": [1, 2, 3] }"); + assertEquals(Arrays.asList(1, 2, 3), converter.toCopycatData(arrayJson)); + } + + + @Test + public void booleanToJson() { + JsonNode converted = converter.fromCopycatData(true); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"boolean\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(true, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).booleanValue()); + } + + @Test + public void intToJson() { + JsonNode converted = converter.fromCopycatData(12); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"int\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(12, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).intValue()); + } + + @Test + public void longToJson() { + JsonNode converted = converter.fromCopycatData(4398046511104L); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"long\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(4398046511104L, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).longValue()); + } + + @Test + public void floatToJson() { + JsonNode converted = converter.fromCopycatData(12.34f); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"float\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(12.34f, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).floatValue(), 0.001); + } + + @Test + public void doubleToJson() { + JsonNode converted = converter.fromCopycatData(12.34); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"double\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(12.34, converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).doubleValue(), 0.001); + } + + @Test + public void bytesToJson() throws IOException { + JsonNode converted = converter.fromCopycatData("test-string".getBytes()); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"bytes\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(ByteBuffer.wrap("test-string".getBytes()), + ByteBuffer.wrap(converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).binaryValue())); + } + + @Test + public void stringToJson() { + JsonNode converted = converter.fromCopycatData("test-string"); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"string\" }"), converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals("test-string", converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME).textValue()); + } + + @Test + public void arrayToJson() { + JsonNode converted = converter.fromCopycatData(Arrays.asList(1, 2, 3)); + validateEnvelope(converted); + assertEquals(parse("{ \"type\": \"array\", \"items\": { \"type\": \"int\" } }"), + converted.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertEquals(JsonNodeFactory.instance.arrayNode().add(1).add(2).add(3), + converted.get(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME)); + } + + + private JsonNode parse(String json) { + try { + return objectMapper.readTree(json); + } catch (IOException e) { + fail("IOException during JSON parse: " + e.getMessage()); + throw new RuntimeException("failed"); + } + } + + private void validateEnvelope(JsonNode env) { + assertNotNull(env); + assertTrue(env.isObject()); + assertEquals(2, env.size()); + assertTrue(env.has(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME)); + assertTrue(env.get(JsonSchema.ENVELOPE_SCHEMA_FIELD_NAME).isObject()); + assertTrue(env.has(JsonSchema.ENVELOPE_PAYLOAD_FIELD_NAME)); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.java ---------------------------------------------------------------------- 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 new file mode 100644 index 0000000..130a529 --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/CopycatStandalone.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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ + +package org.apache.kafka.copycat.cli; + +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.utils.Utils; +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.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; + +/** + * <p> + * Command line utility that runs Copycat as a standalone process. In this mode, work is not + * distributed. Instead, all the normal Copycat machinery works within a single process. This is + * useful for ad hoc, small, or experimental jobs. + * </p> + * <p> + * By default, no job configs or offset data is persistent. You can make jobs persistent and + * fault tolerant by overriding the settings to use file storage for both. + * </p> + */ +@InterfaceStability.Unstable +public class CopycatStandalone { + private static final Logger log = LoggerFactory.getLogger(CopycatStandalone.class); + + public static void main(String[] args) throws Exception { + Properties workerProps; + Properties connectorProps; + + if (args.length < 2) { + log.info("Usage: CopycatStandalone 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); + Herder herder = new StandaloneHerder(worker); + final org.apache.kafka.copycat.runtime.Copycat copycat = new org.apache.kafka.copycat.runtime.Copycat(worker, herder); + copycat.start(); + + try { + for (final String connectorPropsFile : Arrays.copyOfRange(args, 1, args.length)) { + connectorProps = Utils.loadProps(connectorPropsFile); + FutureCallback<String> cb = new FutureCallback<>(new Callback<String>() { + @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(); + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java ---------------------------------------------------------------------- diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java new file mode 100644 index 0000000..46229db --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/cli/WorkerConfig.java @@ -0,0 +1,141 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ + +package org.apache.kafka.copycat.cli; + +import org.apache.kafka.common.annotation.InterfaceStability; +import org.apache.kafka.common.config.AbstractConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigDef.Importance; +import org.apache.kafka.common.config.ConfigDef.Type; + +import java.util.Properties; +import java.util.Set; + +/** + * Configuration for standalone workers. + */ +@InterfaceStability.Unstable +public class WorkerConfig extends AbstractConfig { + + public static final String CLUSTER_CONFIG = "cluster"; + private static final String + CLUSTER_CONFIG_DOC = + "ID for this cluster, which is used to provide a namespace so multiple Copycat clusters " + + "or instances may co-exist while sharing a single Kafka cluster."; + public static final String CLUSTER_DEFAULT = "copycat"; + + public static final String BOOTSTRAP_SERVERS_CONFIG = "bootstrap.servers"; + public static final String BOOTSTRAP_SERVERS_DOC + = "A list of host/port pairs to use for establishing the initial connection to the Kafka " + + "cluster. The client will make use of all servers irrespective of which servers are " + + "specified here for bootstrapping—this list only impacts the initial hosts used " + + "to discover the full set of servers. This list should be in the form " + + "<code>host1:port1,host2:port2,...</code>. Since these servers are just used for the " + + "initial connection to discover the full cluster membership (which may change " + + "dynamically), this list need not contain the full set of servers (you may want more " + + "than one, though, in case a server is down)."; + public static final String BOOTSTRAP_SERVERS_DEFAULT = "localhost:9092"; + + public static final String KEY_CONVERTER_CLASS_CONFIG = "key.converter"; + public static final String KEY_CONVERTER_CLASS_DOC = + "Converter class for key Copycat data that implements the <code>Converter</code> interface."; + + public static final String VALUE_CONVERTER_CLASS_CONFIG = "value.converter"; + public static final String VALUE_CONVERTER_CLASS_DOC = + "Converter class for value Copycat data that implements the <code>Converter</code> interface."; + + public static final String KEY_SERIALIZER_CLASS_CONFIG = "key.serializer"; + public static final String KEY_SERIALIZER_CLASS_DOC = + "Serializer class for key that implements the <code>Serializer</code> interface."; + + public static final String VALUE_SERIALIZER_CLASS_CONFIG = "value.serializer"; + public static final String VALUE_SERIALIZER_CLASS_DOC = + "Serializer class for value that implements the <code>Serializer</code> interface."; + + public static final String KEY_DESERIALIZER_CLASS_CONFIG = "key.deserializer"; + public static final String KEY_DESERIALIZER_CLASS_DOC = + "Serializer class for key that implements the <code>Deserializer</code> interface."; + + public static final String VALUE_DESERIALIZER_CLASS_CONFIG = "value.deserializer"; + public static final String VALUE_DESERIALIZER_CLASS_DOC = + "Deserializer class for value that implements the <code>Deserializer</code> interface."; + + public static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG + = "task.shutdown.graceful.timeout.ms"; + private static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC = + "Amount of time to wait for tasks to shutdown gracefully. This is the total amount of time," + + " not per task. All task have shutdown triggered, then they are waited on sequentially."; + private static final String TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DEFAULT = "5000"; + + public static final String OFFSET_COMMIT_INTERVAL_MS_CONFIG = "offset.flush.interval.ms"; + private static final String OFFSET_COMMIT_INTERVAL_MS_DOC + = "Interval at which to try committing offsets for tasks."; + public static final long OFFSET_COMMIT_INTERVAL_MS_DEFAULT = 60000L; + + public static final String OFFSET_COMMIT_TIMEOUT_MS_CONFIG = "offset.flush.timeout.ms"; + private static final String OFFSET_COMMIT_TIMEOUT_MS_DOC + = "Maximum number of milliseconds to wait for records to flush and partition offset data to be" + + " committed to offset storage before cancelling the process and restoring the offset " + + "data to be committed in a future attempt."; + public static final long OFFSET_COMMIT_TIMEOUT_MS_DEFAULT = 5000L; + + private static ConfigDef config; + + static { + config = new ConfigDef() + .define(CLUSTER_CONFIG, Type.STRING, CLUSTER_DEFAULT, Importance.HIGH, CLUSTER_CONFIG_DOC) + .define(BOOTSTRAP_SERVERS_CONFIG, Type.LIST, BOOTSTRAP_SERVERS_DEFAULT, + Importance.HIGH, BOOTSTRAP_SERVERS_DOC) + .define(KEY_CONVERTER_CLASS_CONFIG, Type.CLASS, + Importance.HIGH, KEY_CONVERTER_CLASS_DOC) + .define(VALUE_CONVERTER_CLASS_CONFIG, Type.CLASS, + Importance.HIGH, VALUE_CONVERTER_CLASS_DOC) + .define(KEY_SERIALIZER_CLASS_CONFIG, Type.CLASS, + Importance.HIGH, KEY_SERIALIZER_CLASS_DOC) + .define(VALUE_SERIALIZER_CLASS_CONFIG, Type.CLASS, + Importance.HIGH, VALUE_SERIALIZER_CLASS_DOC) + .define(KEY_DESERIALIZER_CLASS_CONFIG, Type.CLASS, + Importance.HIGH, KEY_DESERIALIZER_CLASS_DOC) + .define(VALUE_DESERIALIZER_CLASS_CONFIG, Type.CLASS, + Importance.HIGH, VALUE_DESERIALIZER_CLASS_DOC) + .define(TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_CONFIG, Type.LONG, + TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DEFAULT, Importance.LOW, + TASK_SHUTDOWN_GRACEFUL_TIMEOUT_MS_DOC) + .define(OFFSET_COMMIT_INTERVAL_MS_CONFIG, Type.LONG, OFFSET_COMMIT_INTERVAL_MS_DEFAULT, + Importance.LOW, OFFSET_COMMIT_INTERVAL_MS_DOC) + .define(OFFSET_COMMIT_TIMEOUT_MS_CONFIG, Type.LONG, OFFSET_COMMIT_TIMEOUT_MS_DEFAULT, + Importance.LOW, OFFSET_COMMIT_TIMEOUT_MS_DOC); + } + + public WorkerConfig() { + this(new Properties()); + } + + public WorkerConfig(Properties props) { + super(config, props); + } + + public Properties getUnusedProperties() { + Set<String> unusedKeys = this.unused(); + Properties unusedProps = new Properties(); + for (String key : unusedKeys) { + unusedProps.put(key, this.originals().get(key)); + } + return unusedProps; + } +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/ConnectorConfig.java ---------------------------------------------------------------------- diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/ConnectorConfig.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/ConnectorConfig.java new file mode 100644 index 0000000..e3fcc1c --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/ConnectorConfig.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 + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ + +package org.apache.kafka.copycat.runtime; + +import org.apache.kafka.common.config.AbstractConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigDef.Importance; +import org.apache.kafka.common.config.ConfigDef.Type; + +import java.util.Properties; +import java.util.Set; + +/** + * <p> + * Configuration options for Connectors. These only include Copycat system-level configuration + * options (e.g. Connector class name, timeouts used by Copycat to control the connector) but does + * not include Connector-specific options (e.g. database connection settings). + * </p> + * <p> + * Note that some of these options are not required for all connectors. For example TOPICS_CONFIG + * is sink-specific. + * </p> + */ +public class ConnectorConfig extends AbstractConfig { + + public static final String NAME_CONFIG = "name"; + private static final String NAME_DOC = "Globally unique name to use for this connector."; + + public static final String CONNECTOR_CLASS_CONFIG = "connector.class"; + private static final String CONNECTOR_CLASS_DOC = + "Name of the class for this connector. Must be a subclass of org.apache.kafka.copycat.connector" + + ".Connector"; + + public static final String TASKS_MAX_CONFIG = "tasks.max"; + private static final String TASKS_MAX_DOC = "Maximum number of tasks to use for this connector."; + public static final int TASKS_MAX_DEFAULT = 1; + + public static final String TOPICS_CONFIG = "topics"; + private static final String TOPICS_DOC = ""; + public static final String TOPICS_DEFAULT = ""; + + private static ConfigDef config; + + static { + config = new ConfigDef() + .define(NAME_CONFIG, Type.STRING, Importance.HIGH, NAME_DOC) + .define(CONNECTOR_CLASS_CONFIG, Type.STRING, Importance.HIGH, CONNECTOR_CLASS_DOC) + .define(TASKS_MAX_CONFIG, Type.INT, TASKS_MAX_DEFAULT, Importance.HIGH, TASKS_MAX_DOC) + .define(TOPICS_CONFIG, Type.LIST, TOPICS_DEFAULT, Importance.HIGH, TOPICS_DOC); + } + + private Properties originalProperties; + + public ConnectorConfig() { + this(new Properties()); + } + + public ConnectorConfig(Properties props) { + super(config, props); + this.originalProperties = props; + } + + public Properties getUnusedProperties() { + Set<String> unusedKeys = this.unused(); + Properties unusedProps = new Properties(); + for (String key : unusedKeys) { + unusedProps.setProperty(key, originalProperties.getProperty(key)); + } + return unusedProps; + } + +} http://git-wip-us.apache.org/repos/asf/kafka/blob/f6acfb08/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java ---------------------------------------------------------------------- diff --git a/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java new file mode 100644 index 0000000..e8dfe14 --- /dev/null +++ b/copycat/runtime/src/main/java/org/apache/kafka/copycat/runtime/Copycat.java @@ -0,0 +1,94 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * <p/> + * http://www.apache.org/licenses/LICENSE-2.0 + * <p/> + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + **/ + +package org.apache.kafka.copycat.runtime; + +import org.apache.kafka.common.annotation.InterfaceStability; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * This class ties together all the components of a Copycat process (herder, worker, + * storage, command interface), managing their lifecycle. + */ +@InterfaceStability.Unstable +public class Copycat { + private static final Logger log = LoggerFactory.getLogger(Copycat.class); + + private final Worker worker; + private final Herder herder; + private final CountDownLatch startLatch = new CountDownLatch(1); + private final CountDownLatch stopLatch = new CountDownLatch(1); + private final AtomicBoolean shutdown = new AtomicBoolean(false); + private final ShutdownHook shutdownHook; + + public Copycat(Worker worker, Herder herder) { + log.debug("Copycat created"); + this.worker = worker; + this.herder = herder; + shutdownHook = new ShutdownHook(); + } + + public void start() { + log.info("Copycat starting"); + Runtime.getRuntime().addShutdownHook(shutdownHook); + + worker.start(); + herder.start(); + + log.info("Copycat started"); + + startLatch.countDown(); + } + + public void stop() { + boolean wasShuttingDown = shutdown.getAndSet(true); + if (!wasShuttingDown) { + log.info("Copycat stopping"); + + herder.stop(); + worker.stop(); + + log.info("Copycat stopped"); + } + + stopLatch.countDown(); + } + + public void awaitStop() { + try { + stopLatch.await(); + } catch (InterruptedException e) { + log.error("Interrupted waiting for Copycat to shutdown"); + } + } + + private class ShutdownHook extends Thread { + @Override + public void run() { + try { + startLatch.await(); + Copycat.this.stop(); + } catch (InterruptedException e) { + log.error("Interrupted in shutdown hook while waiting for copycat startup to finish"); + } + } + } +}
