You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2017/12/06 15:53:22 UTC
[1/2] nifi git commit: NIFI-4664, NIFI-4662, NIFI-4660,
NIFI-4659 moved tests which are timing/threading/network dependent
and brittle to integration tests and un-ignored tests that are IT. Updated
travis to reduce impact on infra and appveyor now skips
Repository: nifi
Updated Branches:
refs/heads/master c730f802b -> cdc1facf3
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
new file mode 100644
index 0000000..69956e7
--- /dev/null
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
@@ -0,0 +1,135 @@
+/*
+ * 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.nifi.processors.kafka.pubsub;
+
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ITConsumeKafka {
+
+ ConsumerLease mockLease = null;
+ ConsumerPool mockConsumerPool = null;
+
+ @Before
+ public void setup() {
+ mockLease = mock(ConsumerLease.class);
+ mockConsumerPool = mock(ConsumerPool.class);
+ }
+
+ @Test
+ public void validateGetAllMessages() throws Exception {
+ String groupName = "validateGetAllMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka_1_0 proc = new ConsumeKafka_1_0() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_1_0.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka_1_0.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_1_0.AUTO_OFFSET_RESET, ConsumeKafka_1_0.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetAllMessagesPattern() throws Exception {
+ String groupName = "validateGetAllMessagesPattern";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka_1_0 proc = new ConsumeKafka_1_0() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_1_0.TOPICS, "(fo.*)|(ba)");
+ runner.setProperty(ConsumeKafka_1_0.TOPIC_TYPE, "pattern");
+ runner.setProperty(ConsumeKafka_1_0.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_1_0.AUTO_OFFSET_RESET, ConsumeKafka_1_0.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetErrorMessages() throws Exception {
+ String groupName = "validateGetErrorMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(true, false);
+ when(mockLease.commit()).thenReturn(Boolean.FALSE);
+
+ ConsumeKafka_1_0 proc = new ConsumeKafka_1_0() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_1_0.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka_1_0.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_1_0.AUTO_OFFSET_RESET, ConsumeKafka_1_0.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(2)).continuePolling();
+ verify(mockLease, times(1)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/ITLumberjackSocketChannelHandler.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/ITLumberjackSocketChannelHandler.java b/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/ITLumberjackSocketChannelHandler.java
new file mode 100644
index 0000000..7d54180
--- /dev/null
+++ b/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/ITLumberjackSocketChannelHandler.java
@@ -0,0 +1,207 @@
+/*
+ * 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.nifi.processors.lumberjack.handler;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import javax.net.ssl.SSLContext;
+import javax.xml.bind.DatatypeConverter;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.util.listen.dispatcher.AsyncChannelDispatcher;
+import org.apache.nifi.processor.util.listen.dispatcher.ChannelDispatcher;
+import org.apache.nifi.processor.util.listen.dispatcher.SocketChannelDispatcher;
+import org.apache.nifi.processor.util.listen.event.Event;
+import org.apache.nifi.processor.util.listen.event.EventFactory;
+import org.apache.nifi.processor.util.listen.handler.ChannelHandlerFactory;
+import org.apache.nifi.processor.util.listen.response.ChannelResponder;
+import org.apache.nifi.processors.lumberjack.event.LumberjackMetadata;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+
+
+
+public class ITLumberjackSocketChannelHandler {
+ private EventFactory<TestEvent> eventFactory;
+ private ChannelHandlerFactory<TestEvent,AsyncChannelDispatcher> channelHandlerFactory;
+ private BlockingQueue<ByteBuffer> byteBuffers;
+ private BlockingQueue<TestEvent> events;
+ private ComponentLog logger = Mockito.mock(ComponentLog.class);
+ private int maxConnections;
+ private SSLContext sslContext;
+ private Charset charset;
+ private ChannelDispatcher dispatcher;
+
+ @Before
+ public void setup() {
+ eventFactory = new TestEventHolderFactory();
+ channelHandlerFactory = new LumberjackSocketChannelHandlerFactory<>();
+
+ byteBuffers = new LinkedBlockingQueue<>();
+ byteBuffers.add(ByteBuffer.allocate(4096));
+
+ events = new LinkedBlockingQueue<>();
+ logger = Mockito.mock(ComponentLog.class);
+
+ maxConnections = 1;
+ sslContext = null;
+ charset = StandardCharsets.UTF_8;
+
+ dispatcher = new SocketChannelDispatcher<>(eventFactory, channelHandlerFactory, byteBuffers, events, logger,
+ maxConnections, sslContext, charset);
+
+ }
+
+ @Test
+ public void testBasicHandling() throws IOException, InterruptedException {
+ final String multiFrameData = "3143000000d7785ec48fcf6ac4201087b3bbe9defb06be40ab669b1602bdf5d49728" +
+ "031957a97f82232979fbaaa7c0924b2e018701f537f37df2ab699a53aea75cad321673ffe43a38e4e04c043f02" +
+ "1f71461b26873e711bee9480f48b0af10fe2889113b8c9e28f4322b82395413a50cafd79957c253d0b992faf41" +
+ "29c2f27c12e5af35be2cedbec133d9b34e0ee27db87db05596fd62f4680796b421964fc9b032ac4dcb54d2575" +
+ "a28a3559df3413ae7be12edf6e9367c2e07f95ca4a848bb856e1b42ed61427d45da2df4f628f40f0000ffff01000" +
+ "0ffff35e0eff0";
+ final List<String> messages = new ArrayList<>();
+ messages.add(multiFrameData);
+
+ run(messages);
+
+ // Check for the 4 frames (from the hex string above) are back...
+ Assert.assertEquals(4, events.size());
+
+ boolean found1 = false;
+ boolean found2 = false;
+ boolean found3 = false;
+ boolean found4 = false;
+
+ TestEvent event;
+ while((event = events.poll()) != null) {
+ Map<String,String> metadata = event.metadata;
+ Assert.assertTrue(metadata.containsKey(LumberjackMetadata.SEQNUMBER_KEY));
+
+ final String seqNum = metadata.get(LumberjackMetadata.SEQNUMBER_KEY);
+ if (seqNum.equals("1")) {
+ found1 = true;
+ } else if (seqNum.equals("2")) {
+ found2 = true;
+ } else if (seqNum.equals("3")) {
+ found3 = true;
+ } else if (seqNum.equals("4")) {
+ found4 = true;
+ }
+ }
+
+ Assert.assertTrue(found1);
+ Assert.assertTrue(found2);
+ Assert.assertTrue(found3);
+ Assert.assertTrue(found4);
+ }
+
+ protected void run(List<String> messages) throws IOException, InterruptedException {
+ final ByteBuffer buffer = ByteBuffer.allocate(1024);
+ try {
+ // starts the dispatcher listening on port 0 so it selects a random port
+ dispatcher.open(null, 0, 4096);
+
+ // starts a thread to run the dispatcher which will accept/read connections
+ Thread dispatcherThread = new Thread(dispatcher);
+ dispatcherThread.start();
+
+
+ // create a client connection to the port the dispatcher is listening on
+ final int realPort = dispatcher.getPort();
+ try (SocketChannel channel = SocketChannel.open()) {
+ channel.connect(new InetSocketAddress("localhost", realPort));
+ Thread.sleep(100);
+
+ // send the provided messages
+ for (int i=0; i < messages.size(); i++) {
+ buffer.clear();
+ buffer.put(DatatypeConverter.parseHexBinary(messages.get(i)));
+ buffer.flip();
+
+ while (buffer.hasRemaining()) {
+ channel.write(buffer);
+ }
+ Thread.sleep(1);
+ }
+ }
+
+ // wait up to 10 seconds to verify the responses
+ long timeout = 10000;
+ long startTime = System.currentTimeMillis();
+ while (events.size() < messages.size() && (System.currentTimeMillis() - startTime < timeout)) {
+ Thread.sleep(100);
+ }
+
+ // should have gotten an event for each message sent
+ Assert.assertEquals(4, events.size());
+
+ } finally {
+ // stop the dispatcher thread and ensure we shut down handler threads
+ dispatcher.close();
+ }
+ }
+
+ // Test event to produce from the data
+ private static class TestEvent implements Event<SocketChannel> {
+
+ private byte[] data;
+ private Map<String, String> metadata;
+
+ public TestEvent(byte[] data, Map<String, String> metadata) {
+ this.data = data;
+ this.metadata = metadata;
+ }
+
+ @Override
+ public String getSender() {
+ return metadata.get(EventFactory.SENDER_KEY);
+ }
+
+ @Override
+ public byte[] getData() {
+ return data;
+ }
+
+ @Override
+ public ChannelResponder<SocketChannel> getResponder() {
+ return null;
+ }
+ }
+
+ // Factory to create test events and send responses for testing
+ private static class TestEventHolderFactory implements EventFactory<TestEvent> {
+
+ @Override
+ public TestEvent create(final byte[] data, final Map<String, String> metadata, final ChannelResponder responder) {
+ return new TestEvent(data, metadata);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/TestLumberjackSocketChannelHandler.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/TestLumberjackSocketChannelHandler.java b/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/TestLumberjackSocketChannelHandler.java
deleted file mode 100644
index ee5a040..0000000
--- a/nifi-nar-bundles/nifi-lumberjack-bundle/nifi-lumberjack-processors/src/test/java/org/apache/nifi/processors/lumberjack/handler/TestLumberjackSocketChannelHandler.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.lumberjack.handler;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.channels.SocketChannel;
-import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-
-import javax.net.ssl.SSLContext;
-import javax.xml.bind.DatatypeConverter;
-
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.util.listen.dispatcher.AsyncChannelDispatcher;
-import org.apache.nifi.processor.util.listen.dispatcher.ChannelDispatcher;
-import org.apache.nifi.processor.util.listen.dispatcher.SocketChannelDispatcher;
-import org.apache.nifi.processor.util.listen.event.Event;
-import org.apache.nifi.processor.util.listen.event.EventFactory;
-import org.apache.nifi.processor.util.listen.handler.ChannelHandlerFactory;
-import org.apache.nifi.processor.util.listen.response.ChannelResponder;
-import org.apache.nifi.processors.lumberjack.event.LumberjackMetadata;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mockito;
-
-
-
-
-public class TestLumberjackSocketChannelHandler {
- private EventFactory<TestEvent> eventFactory;
- private ChannelHandlerFactory<TestEvent,AsyncChannelDispatcher> channelHandlerFactory;
- private BlockingQueue<ByteBuffer> byteBuffers;
- private BlockingQueue<TestEvent> events;
- private ComponentLog logger = Mockito.mock(ComponentLog.class);
- private int maxConnections;
- private SSLContext sslContext;
- private Charset charset;
- private ChannelDispatcher dispatcher;
-
- @Before
- public void setup() {
- eventFactory = new TestEventHolderFactory();
- channelHandlerFactory = new LumberjackSocketChannelHandlerFactory<>();
-
- byteBuffers = new LinkedBlockingQueue<>();
- byteBuffers.add(ByteBuffer.allocate(4096));
-
- events = new LinkedBlockingQueue<>();
- logger = Mockito.mock(ComponentLog.class);
-
- maxConnections = 1;
- sslContext = null;
- charset = StandardCharsets.UTF_8;
-
- dispatcher = new SocketChannelDispatcher<>(eventFactory, channelHandlerFactory, byteBuffers, events, logger,
- maxConnections, sslContext, charset);
-
- }
-
- @Test
- public void testBasicHandling() throws IOException, InterruptedException {
- final String multiFrameData = "3143000000d7785ec48fcf6ac4201087b3bbe9defb06be40ab669b1602bdf5d49728" +
- "031957a97f82232979fbaaa7c0924b2e018701f537f37df2ab699a53aea75cad321673ffe43a38e4e04c043f02" +
- "1f71461b26873e711bee9480f48b0af10fe2889113b8c9e28f4322b82395413a50cafd79957c253d0b992faf41" +
- "29c2f27c12e5af35be2cedbec133d9b34e0ee27db87db05596fd62f4680796b421964fc9b032ac4dcb54d2575" +
- "a28a3559df3413ae7be12edf6e9367c2e07f95ca4a848bb856e1b42ed61427d45da2df4f628f40f0000ffff01000" +
- "0ffff35e0eff0";
- final List<String> messages = new ArrayList<>();
- messages.add(multiFrameData);
-
- run(messages);
-
- // Check for the 4 frames (from the hex string above) are back...
- Assert.assertEquals(4, events.size());
-
- boolean found1 = false;
- boolean found2 = false;
- boolean found3 = false;
- boolean found4 = false;
-
- TestEvent event;
- while((event = events.poll()) != null) {
- Map<String,String> metadata = event.metadata;
- Assert.assertTrue(metadata.containsKey(LumberjackMetadata.SEQNUMBER_KEY));
-
- final String seqNum = metadata.get(LumberjackMetadata.SEQNUMBER_KEY);
- if (seqNum.equals("1")) {
- found1 = true;
- } else if (seqNum.equals("2")) {
- found2 = true;
- } else if (seqNum.equals("3")) {
- found3 = true;
- } else if (seqNum.equals("4")) {
- found4 = true;
- }
- }
-
- Assert.assertTrue(found1);
- Assert.assertTrue(found2);
- Assert.assertTrue(found3);
- Assert.assertTrue(found4);
- }
-
- protected void run(List<String> messages) throws IOException, InterruptedException {
- final ByteBuffer buffer = ByteBuffer.allocate(1024);
- try {
- // starts the dispatcher listening on port 0 so it selects a random port
- dispatcher.open(null, 0, 4096);
-
- // starts a thread to run the dispatcher which will accept/read connections
- Thread dispatcherThread = new Thread(dispatcher);
- dispatcherThread.start();
-
-
- // create a client connection to the port the dispatcher is listening on
- final int realPort = dispatcher.getPort();
- try (SocketChannel channel = SocketChannel.open()) {
- channel.connect(new InetSocketAddress("localhost", realPort));
- Thread.sleep(100);
-
- // send the provided messages
- for (int i=0; i < messages.size(); i++) {
- buffer.clear();
- buffer.put(DatatypeConverter.parseHexBinary(messages.get(i)));
- buffer.flip();
-
- while (buffer.hasRemaining()) {
- channel.write(buffer);
- }
- Thread.sleep(1);
- }
- }
-
- // wait up to 10 seconds to verify the responses
- long timeout = 10000;
- long startTime = System.currentTimeMillis();
- while (events.size() < messages.size() && (System.currentTimeMillis() - startTime < timeout)) {
- Thread.sleep(100);
- }
-
- // should have gotten an event for each message sent
- Assert.assertEquals(4, events.size());
-
- } finally {
- // stop the dispatcher thread and ensure we shut down handler threads
- dispatcher.close();
- }
- }
-
- // Test event to produce from the data
- private static class TestEvent implements Event<SocketChannel> {
-
- private byte[] data;
- private Map<String, String> metadata;
-
- public TestEvent(byte[] data, Map<String, String> metadata) {
- this.data = data;
- this.metadata = metadata;
- }
-
- @Override
- public String getSender() {
- return metadata.get(EventFactory.SENDER_KEY);
- }
-
- @Override
- public byte[] getData() {
- return data;
- }
-
- @Override
- public ChannelResponder<SocketChannel> getResponder() {
- return null;
- }
- }
-
- // Factory to create test events and send responses for testing
- private static class TestEventHolderFactory implements EventFactory<TestEvent> {
-
- @Override
- public TestEvent create(final byte[] data, final Map<String, String> metadata, final ChannelResponder responder) {
- return new TestEvent(data, metadata);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ITListenSyslogGroovy.groovy
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ITListenSyslogGroovy.groovy b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ITListenSyslogGroovy.groovy
new file mode 100644
index 0000000..1f44225
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ITListenSyslogGroovy.groovy
@@ -0,0 +1,111 @@
+/*
+ * 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.nifi.processors.standard
+
+import org.apache.nifi.processor.ProcessContext
+import org.apache.nifi.processor.ProcessSessionFactory
+import org.apache.nifi.processors.standard.syslog.SyslogParser
+import org.apache.nifi.util.TestRunner
+import org.apache.nifi.util.TestRunners
+import org.bouncycastle.util.encoders.Hex
+import org.junit.After
+import org.junit.Assert
+import org.junit.Before
+import org.junit.BeforeClass
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.JUnit4
+import org.slf4j.Logger
+import org.slf4j.LoggerFactory
+
+@RunWith(JUnit4.class)
+class ListenSyslogGroovyTest extends GroovyTestCase {
+ private static final Logger logger = LoggerFactory.getLogger(ListenSyslogGroovyTest.class)
+
+ static final String ZERO_LENGTH_MESSAGE = " \n"
+
+ @BeforeClass
+ static void setUpOnce() throws Exception {
+ logger.metaClass.methodMissing = { String name, args ->
+ logger.info("[${name?.toUpperCase()}] ${(args as List).join(" ")}")
+ }
+ }
+
+ @Before
+ void setUp() throws Exception {
+ }
+
+ @After
+ void tearDown() throws Exception {
+ }
+
+ @Test
+ void testShouldHandleZeroLengthUDP() throws Exception {
+ // Arrange
+ final ListenSyslog proc = new ListenSyslog()
+ final TestRunner runner = TestRunners.newTestRunner(proc)
+ runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue())
+ runner.setProperty(ListenSyslog.PORT, "0")
+
+ // schedule to start listening on a random port
+ final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory()
+ final ProcessContext context = runner.getProcessContext()
+ proc.onScheduled(context)
+
+ // Inject a SyslogParser which will always return null
+ def nullEventParser = [parseEvent: { byte[] bytes, String sender ->
+ logger.mock("Regardless of input bytes: [${Hex.toHexString(bytes)}] and sender: [${sender}], this parser will return null")
+ return null
+ }] as SyslogParser
+ proc.parser = nullEventParser
+
+ final int numMessages = 10
+ final int port = proc.getPort()
+ Assert.assertTrue(port > 0)
+
+ // write some TCP messages to the port in the background
+ final Thread sender = new Thread(new ITListenSyslog.SingleConnectionSocketSender(port, numMessages, 100, ZERO_LENGTH_MESSAGE))
+ sender.setDaemon(true)
+ sender.start()
+
+ // Act
+
+ // call onTrigger until we read all messages, or 30 seconds passed
+ try {
+ int numFailed = 0
+ long timeout = System.currentTimeMillis() + 30000
+
+ while (numFailed < numMessages && System.currentTimeMillis() < timeout) {
+ Thread.sleep(50)
+ proc.onTrigger(context, processSessionFactory)
+ numFailed = runner.getFlowFilesForRelationship(ListenSyslog.REL_INVALID).size()
+ }
+
+ int numSuccess = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size()
+ logger.info("Transferred " + numSuccess + " to SUCCESS and " + numFailed + " to INVALID")
+
+ // Assert
+
+ // all messages should be transferred to invalid
+ Assert.assertEquals("Did not process all the messages", numMessages, numFailed)
+
+ } finally {
+ // unschedule to close connections
+ proc.onUnscheduled()
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ListenSyslogGroovyTest.groovy
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ListenSyslogGroovyTest.groovy b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ListenSyslogGroovyTest.groovy
deleted file mode 100644
index 1c6b4f8..0000000
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/groovy/org/apache/nifi/processors/standard/ListenSyslogGroovyTest.groovy
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.processors.standard
-
-import org.apache.nifi.processor.ProcessContext
-import org.apache.nifi.processor.ProcessSessionFactory
-import org.apache.nifi.processors.standard.syslog.SyslogParser
-import org.apache.nifi.util.TestRunner
-import org.apache.nifi.util.TestRunners
-import org.bouncycastle.util.encoders.Hex
-import org.junit.After
-import org.junit.Assert
-import org.junit.Before
-import org.junit.BeforeClass
-import org.junit.Test
-import org.junit.runner.RunWith
-import org.junit.runners.JUnit4
-import org.slf4j.Logger
-import org.slf4j.LoggerFactory
-
-@RunWith(JUnit4.class)
-class ListenSyslogGroovyTest extends GroovyTestCase {
- private static final Logger logger = LoggerFactory.getLogger(ListenSyslogGroovyTest.class)
-
- static final String ZERO_LENGTH_MESSAGE = " \n"
-
- @BeforeClass
- static void setUpOnce() throws Exception {
- logger.metaClass.methodMissing = { String name, args ->
- logger.info("[${name?.toUpperCase()}] ${(args as List).join(" ")}")
- }
- }
-
- @Before
- void setUp() throws Exception {
- }
-
- @After
- void tearDown() throws Exception {
- }
-
- @Test
- void testShouldHandleZeroLengthUDP() throws Exception {
- // Arrange
- final ListenSyslog proc = new ListenSyslog()
- final TestRunner runner = TestRunners.newTestRunner(proc)
- runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue())
- runner.setProperty(ListenSyslog.PORT, "0")
-
- // schedule to start listening on a random port
- final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory()
- final ProcessContext context = runner.getProcessContext()
- proc.onScheduled(context)
-
- // Inject a SyslogParser which will always return null
- def nullEventParser = [parseEvent: { byte[] bytes, String sender ->
- logger.mock("Regardless of input bytes: [${Hex.toHexString(bytes)}] and sender: [${sender}], this parser will return null")
- return null
- }] as SyslogParser
- proc.parser = nullEventParser
-
- final int numMessages = 10
- final int port = proc.getPort()
- Assert.assertTrue(port > 0)
-
- // write some TCP messages to the port in the background
- final Thread sender = new Thread(new TestListenSyslog.SingleConnectionSocketSender(port, numMessages, 100, ZERO_LENGTH_MESSAGE))
- sender.setDaemon(true)
- sender.start()
-
- // Act
-
- // call onTrigger until we read all messages, or 30 seconds passed
- try {
- int numFailed = 0
- long timeout = System.currentTimeMillis() + 30000
-
- while (numFailed < numMessages && System.currentTimeMillis() < timeout) {
- Thread.sleep(50)
- proc.onTrigger(context, processSessionFactory)
- numFailed = runner.getFlowFilesForRelationship(ListenSyslog.REL_INVALID).size()
- }
-
- int numSuccess = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size()
- logger.info("Transferred " + numSuccess + " to SUCCESS and " + numFailed + " to INVALID")
-
- // Assert
-
- // all messages should be transferred to invalid
- Assert.assertEquals("Did not process all the messages", numMessages, numFailed)
-
- } finally {
- // unschedule to close connections
- proc.onUnscheduled()
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITListenSyslog.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITListenSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITListenSyslog.java
new file mode 100644
index 0000000..a94bb20
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/ITListenSyslog.java
@@ -0,0 +1,402 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processors.standard.syslog.SyslogAttributes;
+import org.apache.nifi.provenance.ProvenanceEventRecord;
+import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.nio.charset.Charset;
+import java.util.List;
+import org.apache.nifi.processors.standard.TestListenSyslog.DatagramSender;
+
+public class ITListenSyslog {
+
+ static final Logger LOGGER = LoggerFactory.getLogger(ITListenSyslog.class);
+
+ static final String PRI = "34";
+ static final String SEV = "2";
+ static final String FAC = "4";
+ static final String TIME = "Oct 13 15:43:23";
+ static final String HOST = "localhost.home";
+ static final String BODY = "some message";
+
+ static final String VALID_MESSAGE = "<" + PRI + ">" + TIME + " " + HOST + " " + BODY;
+ static final String VALID_MESSAGE_TCP = "<" + PRI + ">" + TIME + " " + HOST + " " + BODY + "\n";
+ static final String INVALID_MESSAGE = "this is not valid\n";
+
+ @Test
+ public void testUDP() throws IOException, InterruptedException {
+ final ListenSyslog proc = new ListenSyslog();
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.UDP_VALUE.getValue());
+ runner.setProperty(ListenSyslog.PORT, "0");
+
+ // schedule to start listening on a random port
+ final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
+ final ProcessContext context = runner.getProcessContext();
+ proc.onScheduled(context);
+
+ final int numMessages = 20;
+ final int port = proc.getPort();
+ Assert.assertTrue(port > 0);
+
+ // write some UDP messages to the port in the background
+ final Thread sender = new Thread(new DatagramSender(port, numMessages, 10, VALID_MESSAGE));
+ sender.setDaemon(true);
+ sender.start();
+
+ // call onTrigger until we read all datagrams, or 30 seconds passed
+ try {
+ int numTransferred = 0;
+ long timeout = System.currentTimeMillis() + 30000;
+
+ while (numTransferred < numMessages && System.currentTimeMillis() < timeout) {
+ Thread.sleep(10);
+ proc.onTrigger(context, processSessionFactory);
+ numTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
+ }
+ Assert.assertEquals("Did not process all the datagrams", numMessages, numTransferred);
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
+ checkFlowFile(flowFile, 0, ListenSyslog.UDP_VALUE.getValue());
+
+ final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
+ Assert.assertNotNull(events);
+ Assert.assertEquals(numMessages, events.size());
+
+ final ProvenanceEventRecord event = events.get(0);
+ Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
+ Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("udp"));
+
+ } finally {
+ // unschedule to close connections
+ proc.onUnscheduled();
+ }
+ }
+
+ @Test
+ public void testTCPSingleConnection() throws IOException, InterruptedException {
+ final ListenSyslog proc = new ListenSyslog();
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
+ runner.setProperty(ListenSyslog.PORT, "0");
+
+ // schedule to start listening on a random port
+ final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
+ final ProcessContext context = runner.getProcessContext();
+ proc.onScheduled(context);
+
+ // Allow time for the processor to perform its scheduled start
+ Thread.sleep(500);
+
+ final int numMessages = 20;
+ final int port = proc.getPort();
+ Assert.assertTrue(port > 0);
+
+ // write some TCP messages to the port in the background
+ final Thread sender = new Thread(new SingleConnectionSocketSender(port, numMessages, 10, VALID_MESSAGE_TCP));
+ sender.setDaemon(true);
+ sender.start();
+
+ // call onTrigger until we read all messages, or 30 seconds passed
+ try {
+ int nubTransferred = 0;
+ long timeout = System.currentTimeMillis() + 30000;
+
+ while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
+ Thread.sleep(10);
+ proc.onTrigger(context, processSessionFactory);
+ nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
+ }
+ Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
+ checkFlowFile(flowFile, 0, ListenSyslog.TCP_VALUE.getValue());
+
+ final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
+ Assert.assertNotNull(events);
+ Assert.assertEquals(numMessages, events.size());
+
+ final ProvenanceEventRecord event = events.get(0);
+ Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
+ Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("tcp"));
+
+ } finally {
+ // unschedule to close connections
+ proc.onUnscheduled();
+ }
+ }
+
+ @Test
+ public void testTCPSingleConnectionWithNewLines() throws IOException, InterruptedException {
+ final ListenSyslog proc = new ListenSyslog();
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
+ runner.setProperty(ListenSyslog.PORT, "0");
+
+ // schedule to start listening on a random port
+ final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
+ final ProcessContext context = runner.getProcessContext();
+ proc.onScheduled(context);
+
+ final int numMessages = 3;
+ final int port = proc.getPort();
+ Assert.assertTrue(port > 0);
+
+ // send 3 messages as 1
+ final String multipleMessages = VALID_MESSAGE_TCP + "\n" + VALID_MESSAGE_TCP + "\n" + VALID_MESSAGE_TCP + "\n";
+ final Thread sender = new Thread(new SingleConnectionSocketSender(port, 1, 10, multipleMessages));
+ sender.setDaemon(true);
+ sender.start();
+
+ // call onTrigger until we read all messages, or 30 seconds passed
+ try {
+ int nubTransferred = 0;
+ long timeout = System.currentTimeMillis() + 30000;
+
+ while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
+ Thread.sleep(10);
+ proc.onTrigger(context, processSessionFactory);
+ nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
+ }
+ Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
+ checkFlowFile(flowFile, 0, ListenSyslog.TCP_VALUE.getValue());
+
+ final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
+ Assert.assertNotNull(events);
+ Assert.assertEquals(numMessages, events.size());
+
+ final ProvenanceEventRecord event = events.get(0);
+ Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
+ Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("tcp"));
+
+ } finally {
+ // unschedule to close connections
+ proc.onUnscheduled();
+ }
+ }
+
+ @Test
+ public void testTCPMultipleConnection() throws IOException, InterruptedException {
+ final ListenSyslog proc = new ListenSyslog();
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
+ runner.setProperty(ListenSyslog.MAX_CONNECTIONS, "5");
+ runner.setProperty(ListenSyslog.PORT, "0");
+
+ // schedule to start listening on a random port
+ final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
+ final ProcessContext context = runner.getProcessContext();
+ proc.onScheduled(context);
+
+ final int numMessages = 20;
+ final int port = proc.getPort();
+ Assert.assertTrue(port > 0);
+
+ // write some TCP messages to the port in the background
+ final Thread sender = new Thread(new MultiConnectionSocketSender(port, numMessages, 10, VALID_MESSAGE_TCP));
+ sender.setDaemon(true);
+ sender.start();
+
+ // call onTrigger until we read all messages, or 30 seconds passed
+ try {
+ int nubTransferred = 0;
+ long timeout = System.currentTimeMillis() + 30000;
+
+ while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
+ Thread.sleep(10);
+ proc.onTrigger(context, processSessionFactory);
+ nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
+ }
+ Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
+ checkFlowFile(flowFile, 0, ListenSyslog.TCP_VALUE.getValue());
+
+ final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
+ Assert.assertNotNull(events);
+ Assert.assertEquals(numMessages, events.size());
+
+ final ProvenanceEventRecord event = events.get(0);
+ Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
+ Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("tcp"));
+
+ } finally {
+ // unschedule to close connections
+ proc.onUnscheduled();
+ }
+ }
+
+ @Test
+ public void testInvalid() throws IOException, InterruptedException {
+ final ListenSyslog proc = new ListenSyslog();
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
+ runner.setProperty(ListenSyslog.PORT, "0");
+
+ // schedule to start listening on a random port
+ final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
+ final ProcessContext context = runner.getProcessContext();
+ proc.onScheduled(context);
+
+ final int numMessages = 10;
+ final int port = proc.getPort();
+ Assert.assertTrue(port > 0);
+
+ // write some TCP messages to the port in the background
+ final Thread sender = new Thread(new SingleConnectionSocketSender(port, numMessages, 100, INVALID_MESSAGE));
+ sender.setDaemon(true);
+ sender.start();
+
+ // call onTrigger until we read all messages, or 30 seconds passed
+ try {
+ int nubTransferred = 0;
+ long timeout = System.currentTimeMillis() + 30000;
+
+ while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
+ Thread.sleep(50);
+ proc.onTrigger(context, processSessionFactory);
+ nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_INVALID).size();
+ }
+
+ // all messages should be transferred to invalid
+ Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
+
+ } finally {
+ // unschedule to close connections
+ proc.onUnscheduled();
+ }
+ }
+
+ private void checkFlowFile(final MockFlowFile flowFile, final int port, final String protocol) {
+ flowFile.assertContentEquals(VALID_MESSAGE.replace("\n", ""));
+ Assert.assertEquals(PRI, flowFile.getAttribute(SyslogAttributes.PRIORITY.key()));
+ Assert.assertEquals(SEV, flowFile.getAttribute(SyslogAttributes.SEVERITY.key()));
+ Assert.assertEquals(FAC, flowFile.getAttribute(SyslogAttributes.FACILITY.key()));
+ Assert.assertEquals(TIME, flowFile.getAttribute(SyslogAttributes.TIMESTAMP.key()));
+ Assert.assertEquals(HOST, flowFile.getAttribute(SyslogAttributes.HOSTNAME.key()));
+ Assert.assertEquals(BODY, flowFile.getAttribute(SyslogAttributes.BODY.key()));
+ Assert.assertEquals("true", flowFile.getAttribute(SyslogAttributes.VALID.key()));
+ Assert.assertEquals(String.valueOf(port), flowFile.getAttribute(SyslogAttributes.PORT.key()));
+ Assert.assertEquals(protocol, flowFile.getAttribute(SyslogAttributes.PROTOCOL.key()));
+ Assert.assertTrue(!StringUtils.isBlank(flowFile.getAttribute(SyslogAttributes.SENDER.key())));
+ }
+
+ /**
+ * Sends a given number of datagrams to the given port.
+ */
+ public static final class SingleConnectionSocketSender implements Runnable {
+
+ final int port;
+ final int numMessages;
+ final long delay;
+ final String message;
+
+ public SingleConnectionSocketSender(int port, int numMessages, long delay, String message) {
+ this.port = port;
+ this.numMessages = numMessages;
+ this.delay = delay;
+ this.message = message;
+ }
+
+ @Override
+ public void run() {
+ byte[] bytes = message.getBytes(Charset.forName("UTF-8"));
+ final ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
+
+ try (SocketChannel channel = SocketChannel.open()) {
+ channel.connect(new InetSocketAddress("localhost", port));
+
+ for (int i = 0; i < numMessages; i++) {
+ buffer.clear();
+ buffer.put(bytes);
+ buffer.flip();
+
+ while (buffer.hasRemaining()) {
+ channel.write(buffer);
+ }
+ Thread.sleep(delay);
+ }
+ } catch (IOException e) {
+ LOGGER.error(e.getMessage(), e);
+ } catch (InterruptedException e) {
+ LOGGER.error(e.getMessage(), e);
+ }
+ }
+ }
+
+ /**
+ * Sends a given number of datagrams to the given port.
+ */
+ public static final class MultiConnectionSocketSender implements Runnable {
+
+ final int port;
+ final int numMessages;
+ final long delay;
+ final String message;
+
+ public MultiConnectionSocketSender(int port, int numMessages, long delay, String message) {
+ this.port = port;
+ this.numMessages = numMessages;
+ this.delay = delay;
+ this.message = message;
+ }
+
+ @Override
+ public void run() {
+ byte[] bytes = message.getBytes(Charset.forName("UTF-8"));
+ final ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
+
+ for (int i = 0; i < numMessages; i++) {
+ try (SocketChannel channel = SocketChannel.open()) {
+ channel.connect(new InetSocketAddress("localhost", port));
+
+ buffer.clear();
+ buffer.put(bytes);
+ buffer.flip();
+
+ while (buffer.hasRemaining()) {
+ channel.write(buffer);
+ }
+ Thread.sleep(delay);
+ } catch (IOException e) {
+ LOGGER.error(e.getMessage(), e);
+ } catch (InterruptedException e) {
+ LOGGER.error(e.getMessage(), e);
+ }
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
index f96ff22..2c199c1 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestListenSyslog.java
@@ -41,7 +41,6 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
-import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -62,216 +61,11 @@ public class TestListenSyslog {
static final String HOST = "localhost.home";
static final String BODY = "some message";
- static final String VALID_MESSAGE = "<" + PRI + ">" + TIME + " " + HOST + " " + BODY ;
+ static final String VALID_MESSAGE = "<" + PRI + ">" + TIME + " " + HOST + " " + BODY;
static final String VALID_MESSAGE_TCP = "<" + PRI + ">" + TIME + " " + HOST + " " + BODY + "\n";
static final String INVALID_MESSAGE = "this is not valid\n";
@Test
- public void testUDP() throws IOException, InterruptedException {
- final ListenSyslog proc = new ListenSyslog();
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.UDP_VALUE.getValue());
- runner.setProperty(ListenSyslog.PORT, "0");
-
- // schedule to start listening on a random port
- final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
- final ProcessContext context = runner.getProcessContext();
- proc.onScheduled(context);
-
- final int numMessages = 20;
- final int port = proc.getPort();
- Assert.assertTrue(port > 0);
-
- // write some UDP messages to the port in the background
- final Thread sender = new Thread(new DatagramSender(port, numMessages, 10, VALID_MESSAGE));
- sender.setDaemon(true);
- sender.start();
-
- // call onTrigger until we read all datagrams, or 30 seconds passed
- try {
- int numTransferred = 0;
- long timeout = System.currentTimeMillis() + 30000;
-
- while (numTransferred < numMessages && System.currentTimeMillis() < timeout) {
- Thread.sleep(10);
- proc.onTrigger(context, processSessionFactory);
- numTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
- }
- Assert.assertEquals("Did not process all the datagrams", numMessages, numTransferred);
-
- MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
- checkFlowFile(flowFile, 0, ListenSyslog.UDP_VALUE.getValue());
-
- final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
- Assert.assertNotNull(events);
- Assert.assertEquals(numMessages, events.size());
-
- final ProvenanceEventRecord event = events.get(0);
- Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
- Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("udp"));
-
- } finally {
- // unschedule to close connections
- proc.onUnscheduled();
- }
- }
-
- @Test
- public void testTCPSingleConnection() throws IOException, InterruptedException {
- final ListenSyslog proc = new ListenSyslog();
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
- runner.setProperty(ListenSyslog.PORT, "0");
-
- // schedule to start listening on a random port
- final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
- final ProcessContext context = runner.getProcessContext();
- proc.onScheduled(context);
-
- // Allow time for the processor to perform its scheduled start
- Thread.sleep(500);
-
- final int numMessages = 20;
- final int port = proc.getPort();
- Assert.assertTrue(port > 0);
-
- // write some TCP messages to the port in the background
- final Thread sender = new Thread(new SingleConnectionSocketSender(port, numMessages, 10, VALID_MESSAGE_TCP));
- sender.setDaemon(true);
- sender.start();
-
- // call onTrigger until we read all messages, or 30 seconds passed
- try {
- int nubTransferred = 0;
- long timeout = System.currentTimeMillis() + 30000;
-
- while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
- Thread.sleep(10);
- proc.onTrigger(context, processSessionFactory);
- nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
- }
- Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
-
- MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
- checkFlowFile(flowFile, 0, ListenSyslog.TCP_VALUE.getValue());
-
- final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
- Assert.assertNotNull(events);
- Assert.assertEquals(numMessages, events.size());
-
- final ProvenanceEventRecord event = events.get(0);
- Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
- Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("tcp"));
-
- } finally {
- // unschedule to close connections
- proc.onUnscheduled();
- }
- }
-
- @Test
- public void testTCPSingleConnectionWithNewLines() throws IOException, InterruptedException {
- final ListenSyslog proc = new ListenSyslog();
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
- runner.setProperty(ListenSyslog.PORT, "0");
-
- // schedule to start listening on a random port
- final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
- final ProcessContext context = runner.getProcessContext();
- proc.onScheduled(context);
-
- final int numMessages = 3;
- final int port = proc.getPort();
- Assert.assertTrue(port > 0);
-
- // send 3 messages as 1
- final String multipleMessages = VALID_MESSAGE_TCP + "\n" + VALID_MESSAGE_TCP + "\n" + VALID_MESSAGE_TCP + "\n";
- final Thread sender = new Thread(new SingleConnectionSocketSender(port, 1, 10, multipleMessages));
- sender.setDaemon(true);
- sender.start();
-
- // call onTrigger until we read all messages, or 30 seconds passed
- try {
- int nubTransferred = 0;
- long timeout = System.currentTimeMillis() + 30000;
-
- while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
- Thread.sleep(10);
- proc.onTrigger(context, processSessionFactory);
- nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
- }
- Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
-
- MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
- checkFlowFile(flowFile, 0, ListenSyslog.TCP_VALUE.getValue());
-
- final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
- Assert.assertNotNull(events);
- Assert.assertEquals(numMessages, events.size());
-
- final ProvenanceEventRecord event = events.get(0);
- Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
- Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("tcp"));
-
- } finally {
- // unschedule to close connections
- proc.onUnscheduled();
- }
- }
-
- @Test
- public void testTCPMultipleConnection() throws IOException, InterruptedException {
- final ListenSyslog proc = new ListenSyslog();
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
- runner.setProperty(ListenSyslog.MAX_CONNECTIONS, "5");
- runner.setProperty(ListenSyslog.PORT, "0");
-
- // schedule to start listening on a random port
- final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
- final ProcessContext context = runner.getProcessContext();
- proc.onScheduled(context);
-
- final int numMessages = 20;
- final int port = proc.getPort();
- Assert.assertTrue(port > 0);
-
- // write some TCP messages to the port in the background
- final Thread sender = new Thread(new MultiConnectionSocketSender(port, numMessages, 10, VALID_MESSAGE_TCP));
- sender.setDaemon(true);
- sender.start();
-
- // call onTrigger until we read all messages, or 30 seconds passed
- try {
- int nubTransferred = 0;
- long timeout = System.currentTimeMillis() + 30000;
-
- while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
- Thread.sleep(10);
- proc.onTrigger(context, processSessionFactory);
- nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).size();
- }
- Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
-
- MockFlowFile flowFile = runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0);
- checkFlowFile(flowFile, 0, ListenSyslog.TCP_VALUE.getValue());
-
- final List<ProvenanceEventRecord> events = runner.getProvenanceEvents();
- Assert.assertNotNull(events);
- Assert.assertEquals(numMessages, events.size());
-
- final ProvenanceEventRecord event = events.get(0);
- Assert.assertEquals(ProvenanceEventType.RECEIVE, event.getEventType());
- Assert.assertTrue("transit uri must be set and start with proper protocol", event.getTransitUri().toLowerCase().startsWith("tcp"));
-
- } finally {
- // unschedule to close connections
- proc.onUnscheduled();
- }
- }
-
- @Test
public void testBatching() throws IOException, InterruptedException {
final ListenSyslog proc = new ListenSyslog();
final TestRunner runner = TestRunners.newTestRunner(proc);
@@ -325,47 +119,6 @@ public class TestListenSyslog {
}
@Test
- public void testInvalid() throws IOException, InterruptedException {
- final ListenSyslog proc = new ListenSyslog();
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(ListenSyslog.PROTOCOL, ListenSyslog.TCP_VALUE.getValue());
- runner.setProperty(ListenSyslog.PORT, "0");
-
- // schedule to start listening on a random port
- final ProcessSessionFactory processSessionFactory = runner.getProcessSessionFactory();
- final ProcessContext context = runner.getProcessContext();
- proc.onScheduled(context);
-
- final int numMessages = 10;
- final int port = proc.getPort();
- Assert.assertTrue(port > 0);
-
- // write some TCP messages to the port in the background
- final Thread sender = new Thread(new SingleConnectionSocketSender(port, numMessages, 100, INVALID_MESSAGE));
- sender.setDaemon(true);
- sender.start();
-
- // call onTrigger until we read all messages, or 30 seconds passed
- try {
- int nubTransferred = 0;
- long timeout = System.currentTimeMillis() + 30000;
-
- while (nubTransferred < numMessages && System.currentTimeMillis() < timeout) {
- Thread.sleep(50);
- proc.onTrigger(context, processSessionFactory);
- nubTransferred = runner.getFlowFilesForRelationship(ListenSyslog.REL_INVALID).size();
- }
-
- // all messages should be transferred to invalid
- Assert.assertEquals("Did not process all the messages", numMessages, nubTransferred);
-
- } finally {
- // unschedule to close connections
- proc.onUnscheduled();
- }
- }
-
- @Test
public void testParsingError() throws IOException {
final FailParseProcessor proc = new FailParseProcessor();
final TestRunner runner = TestRunners.newTestRunner(proc);
@@ -431,21 +184,6 @@ public class TestListenSyslog {
runner.getFlowFilesForRelationship(ListenSyslog.REL_SUCCESS).get(0).assertContentEquals(VALID_MESSAGE);
}
-
- private void checkFlowFile(final MockFlowFile flowFile, final int port, final String protocol) {
- flowFile.assertContentEquals(VALID_MESSAGE.replace("\n", ""));
- Assert.assertEquals(PRI, flowFile.getAttribute(SyslogAttributes.PRIORITY.key()));
- Assert.assertEquals(SEV, flowFile.getAttribute(SyslogAttributes.SEVERITY.key()));
- Assert.assertEquals(FAC, flowFile.getAttribute(SyslogAttributes.FACILITY.key()));
- Assert.assertEquals(TIME, flowFile.getAttribute(SyslogAttributes.TIMESTAMP.key()));
- Assert.assertEquals(HOST, flowFile.getAttribute(SyslogAttributes.HOSTNAME.key()));
- Assert.assertEquals(BODY, flowFile.getAttribute(SyslogAttributes.BODY.key()));
- Assert.assertEquals("true", flowFile.getAttribute(SyslogAttributes.VALID.key()));
- Assert.assertEquals(String.valueOf(port), flowFile.getAttribute(SyslogAttributes.PORT.key()));
- Assert.assertEquals(protocol, flowFile.getAttribute(SyslogAttributes.PROTOCOL.key()));
- Assert.assertTrue(!StringUtils.isBlank(flowFile.getAttribute(SyslogAttributes.SENDER.key())));
- }
-
/**
* Sends a given number of datagrams to the given port.
*/
@@ -470,51 +208,7 @@ public class TestListenSyslog {
try (DatagramChannel channel = DatagramChannel.open()) {
channel.connect(new InetSocketAddress("localhost", port));
- for (int i=0; i < numMessages; i++) {
- buffer.clear();
- buffer.put(bytes);
- buffer.flip();
-
- while(buffer.hasRemaining()) {
- channel.write(buffer);
- }
-
- Thread.sleep(delay);
- }
- } catch (IOException e) {
- LOGGER.error(e.getMessage(), e);
- } catch (InterruptedException e) {
- LOGGER.error(e.getMessage(), e);
- }
- }
- }
-
- /**
- * Sends a given number of datagrams to the given port.
- */
- public static final class SingleConnectionSocketSender implements Runnable {
-
- final int port;
- final int numMessages;
- final long delay;
- final String message;
-
- public SingleConnectionSocketSender(int port, int numMessages, long delay, String message) {
- this.port = port;
- this.numMessages = numMessages;
- this.delay = delay;
- this.message = message;
- }
-
- @Override
- public void run() {
- byte[] bytes = message.getBytes(Charset.forName("UTF-8"));
- final ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
-
- try (SocketChannel channel = SocketChannel.open()) {
- channel.connect(new InetSocketAddress("localhost", port));
-
- for (int i=0; i < numMessages; i++) {
+ for (int i = 0; i < numMessages; i++) {
buffer.clear();
buffer.put(bytes);
buffer.flip();
@@ -522,6 +216,7 @@ public class TestListenSyslog {
while (buffer.hasRemaining()) {
channel.write(buffer);
}
+
Thread.sleep(delay);
}
} catch (IOException e) {
@@ -532,51 +227,9 @@ public class TestListenSyslog {
}
}
- /**
- * Sends a given number of datagrams to the given port.
- */
- public static final class MultiConnectionSocketSender implements Runnable {
-
- final int port;
- final int numMessages;
- final long delay;
- final String message;
-
- public MultiConnectionSocketSender(int port, int numMessages, long delay, String message) {
- this.port = port;
- this.numMessages = numMessages;
- this.delay = delay;
- this.message = message;
- }
-
- @Override
- public void run() {
- byte[] bytes = message.getBytes(Charset.forName("UTF-8"));
- final ByteBuffer buffer = ByteBuffer.allocate(bytes.length);
-
- for (int i=0; i < numMessages; i++) {
- try (SocketChannel channel = SocketChannel.open()) {
- channel.connect(new InetSocketAddress("localhost", port));
-
- buffer.clear();
- buffer.put(bytes);
- buffer.flip();
-
- while (buffer.hasRemaining()) {
- channel.write(buffer);
- }
- Thread.sleep(delay);
- } catch (IOException e) {
- LOGGER.error(e.getMessage(), e);
- } catch (InterruptedException e) {
- LOGGER.error(e.getMessage(), e);
- }
- }
- }
- }
-
// A mock version of ListenSyslog that will queue the provided events
private static class FailParseProcessor extends ListenSyslog {
+
@Override
protected SyslogParser getParser() {
return new SyslogParser(StandardCharsets.UTF_8) {
@@ -589,6 +242,7 @@ public class TestListenSyslog {
}
private static class CannedMessageProcessor extends ListenSyslog {
+
private final Iterator<RawSyslogEvent> eventItr;
public CannedMessageProcessor(final List<RawSyslogEvent> events) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index a5f408c..146a3a2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1985,7 +1985,9 @@
<!-- Performs execution of Integration Tests using the Maven
FailSafe Plugin. The view of integration tests in this context are those
tests interfacing with external sources and services requiring additional
- resources or credentials that cannot be explicitly provided. -->
+ resources or credentials that cannot be explicitly provided. Also appropriate
+ for tests which depend on inter-thread and/or network or having timing
+ considerations which could make the tests brittle on various environments.-->
<id>integration-tests</id>
<build>
<plugins>
[2/2] nifi git commit: NIFI-4664, NIFI-4662, NIFI-4660,
NIFI-4659 moved tests which are timing/threading/network dependent
and brittle to integration tests and un-ignored tests that are IT. Updated
travis to reduce impact on infra and appveyor now skips
Posted by mc...@apache.org.
NIFI-4664, NIFI-4662, NIFI-4660, NIFI-4659 moved tests which are timing/threading/network dependent and brittle to integration tests and un-ignored tests that are IT. Updated travis to reduce impact on infra and appveyor now skips test runs so is just to prove build works on windows. This closes #2319
squash
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/cdc1facf
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/cdc1facf
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/cdc1facf
Branch: refs/heads/master
Commit: cdc1facf398e6d99cc746b70350ff754e2a975ad
Parents: c730f80
Author: joewitt <jo...@apache.org>
Authored: Tue Dec 5 14:44:07 2017 -0500
Committer: Matt Gilman <ma...@gmail.com>
Committed: Wed Dec 6 10:53:09 2017 -0500
----------------------------------------------------------------------
.travis.yml | 4 +-
appveyor.yml | 4 +-
.../util/list/ITAbstractListProcessor.java | 426 +++++++++++++++++++
.../util/list/TestAbstractListProcessor.java | 353 +--------------
.../kafka/pubsub/ConsumeKafkaTest.java | 98 -----
.../processors/kafka/pubsub/ITConsumeKafka.java | 135 ++++++
.../kafka/pubsub/ConsumeKafkaTest.java | 98 -----
.../processors/kafka/pubsub/ITConsumeKafka.java | 135 ++++++
.../kafka/pubsub/ConsumeKafkaTest.java | 110 -----
.../processors/kafka/pubsub/ITConsumeKafka.java | 150 +++++++
.../kafka/pubsub/ConsumeKafkaTest.java | 98 -----
.../processors/kafka/pubsub/ITConsumeKafka.java | 135 ++++++
.../ITLumberjackSocketChannelHandler.java | 207 +++++++++
.../TestLumberjackSocketChannelHandler.java | 207 ---------
.../standard/ITListenSyslogGroovy.groovy | 111 +++++
.../standard/ListenSyslogGroovyTest.groovy | 111 -----
.../processors/standard/ITListenSyslog.java | 402 +++++++++++++++++
.../processors/standard/TestListenSyslog.java | 356 +---------------
pom.xml | 4 +-
19 files changed, 1717 insertions(+), 1427 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/.travis.yml
----------------------------------------------------------------------
diff --git a/.travis.yml b/.travis.yml
index d7f0f03..e5b0ccb 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -19,8 +19,6 @@ env:
- USER_LANGUAGE=en USER_REGION=US'
- USER_LANGUAGE=fr USER_REGION=FR'
- USER_LANGUAGE=ja USER_REGION=JP'
- - USER_LANGUAGE=pt USER_REGION=BR'
- - USER_LANGUAGE=default USER_REGION=default
os:
- linux
@@ -54,4 +52,4 @@ install:
# Note: The reason the sed is done as part of script is to ensure the pom hack
# won't affect the 'clean install' above
- bash .travis.sh
- - mvn -T 2C -Pcontrib-check -Ddir-only clean install
\ No newline at end of file
+ - mvn -T 2C -Pcontrib-check -Ddir-only clean install
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/appveyor.yml
----------------------------------------------------------------------
diff --git a/appveyor.yml b/appveyor.yml
index c7aa2f2..3e31c19 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -32,9 +32,7 @@ install:
- cmd: SET MAVEN_OPTS=-XX:MaxPermSize=2g -Xmx4g
- cmd: SET JAVA_OPTS=-XX:MaxPermSize=2g -Xmx4g
build_script:
- - mvn -q clean package --batch-mode -DskipTests
-test_script:
- - mvn -q clean install --batch-mode -Pcontrib-check
+ - mvn clean package --batch-mode -DskipTests -Ddir-only
cache:
- C:\maven\
- C:\Users\appveyor\.m2
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/ITAbstractListProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/ITAbstractListProcessor.java b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/ITAbstractListProcessor.java
new file mode 100644
index 0000000..dcf47c6
--- /dev/null
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/ITAbstractListProcessor.java
@@ -0,0 +1,426 @@
+/*
+ * 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.nifi.processor.util.list;
+
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestWatcher;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.nifi.processor.util.list.AbstractListProcessor.PRECISION_MILLIS;
+import static org.apache.nifi.processor.util.list.AbstractListProcessor.PRECISION_MINUTES;
+import static org.apache.nifi.processor.util.list.AbstractListProcessor.PRECISION_SECONDS;
+import static org.apache.nifi.processor.util.list.AbstractListProcessor.TARGET_SYSTEM_TIMESTAMP_PRECISION;
+import org.apache.nifi.processor.util.list.TestAbstractListProcessor.ConcreteListProcessor;
+import org.apache.nifi.processor.util.list.TestAbstractListProcessor.DistributedCache;
+import static org.junit.Assert.assertEquals;
+
+public class ITAbstractListProcessor {
+
+ /**
+ * @return current timestamp in milliseconds, but truncated at specified
+ * target precision (e.g. SECONDS or MINUTES).
+ */
+ private static long getCurrentTimestampMillis(final TimeUnit targetPrecision) {
+ final long timestampInTargetPrecision = targetPrecision.convert(System.currentTimeMillis(), TimeUnit.MILLISECONDS);
+ return TimeUnit.MILLISECONDS.convert(timestampInTargetPrecision, targetPrecision);
+ }
+
+ private static long getSleepMillis(final TimeUnit targetPrecision) {
+ return AbstractListProcessor.LISTING_LAG_MILLIS.get(targetPrecision) * 2;
+ }
+
+ private static final long DEFAULT_SLEEP_MILLIS = getSleepMillis(TimeUnit.MILLISECONDS);
+
+ private ConcreteListProcessor proc;
+ private TestRunner runner;
+
+ @Rule
+ public TestWatcher dumpState = new ListProcessorTestWatcher(
+ () -> {
+ try {
+ return runner.getStateManager().getState(Scope.LOCAL).toMap();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to retrieve state", e);
+ }
+ },
+ () -> proc.entities,
+ () -> runner.getFlowFilesForRelationship(AbstractListProcessor.REL_SUCCESS).stream().map(m -> (FlowFile) m).collect(Collectors.toList())
+ );
+
+ @Before
+ public void setup() {
+ proc = new ConcreteListProcessor();
+ runner = TestRunners.newTestRunner(proc);
+ }
+
+ @Rule
+ public final TemporaryFolder testFolder = new TemporaryFolder();
+
+ /**
+ * <p>
+ * Ensures that files are listed when those are old enough:
+ * <li>Files with last modified timestamp those are old enough to determine
+ * that those are completely written and no further files are expected to be
+ * added with the same timestamp.</li>
+ * <li>This behavior is expected when a processor is scheduled less
+ * frequently, such as hourly or daily.</li>
+ * </p>
+ */
+ @Test
+ public void testAllExistingEntriesEmittedOnFirstIteration() throws Exception {
+ final long oldTimestamp = System.currentTimeMillis() - getSleepMillis(TimeUnit.MILLISECONDS);
+
+ // These entries have existed before the processor runs at the first time.
+ proc.addEntity("name", "id", oldTimestamp);
+ proc.addEntity("name", "id2", oldTimestamp);
+
+ // First run, the above listed entries should be emitted since it has existed.
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
+ runner.clearTransferState();
+
+ // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
+ Thread.sleep(DEFAULT_SLEEP_MILLIS);
+
+ // Run again without introducing any new entries
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ }
+
+ private void testPreviouslySkippedEntriesEmmitedOnNextIteration(final TimeUnit targetPrecision) throws InterruptedException {
+ runner.run();
+
+ final long initialTimestamp = getCurrentTimestampMillis(targetPrecision);
+
+ setTargetSystemTimestampPrecision(targetPrecision);
+
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ proc.addEntity("name", "id", initialTimestamp);
+ proc.addEntity("name", "id2", initialTimestamp);
+ runner.run();
+
+ // First run, the above listed entries would be skipped to avoid write synchronization issues
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
+ Thread.sleep(getSleepMillis(targetPrecision));
+
+ // Run again without introducing any new entries
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
+ }
+
+ /**
+ * <p>
+ * Ensures that newly created files should wait to confirm there is no more
+ * files created with the same timestamp:
+ * <li>If files have the latest modified timestamp at an iteration, then
+ * those should be postponed to be listed</li>
+ * <li>If those files still are the latest files at the next iteration, then
+ * those should be listed</li>
+ * </p>
+ */
+ @Test
+ public void testPreviouslySkippedEntriesEmittedOnNextIterationMilliPrecision() throws Exception {
+ testPreviouslySkippedEntriesEmmitedOnNextIteration(TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * Same as
+ * {@link #testPreviouslySkippedEntriesEmittedOnNextIterationMilliPrecision()}
+ * but simulates that the target filesystem only provide timestamp precision
+ * in Seconds.
+ */
+ @Test
+ public void testPreviouslySkippedEntriesEmittedOnNextIterationSecondPrecision() throws Exception {
+ testPreviouslySkippedEntriesEmmitedOnNextIteration(TimeUnit.SECONDS);
+ }
+
+ /**
+ * Same as
+ * {@link #testPreviouslySkippedEntriesEmittedOnNextIterationMilliPrecision()}
+ * but simulates that the target filesystem only provide timestamp precision
+ * in Minutes.
+ */
+ @Test
+ public void testPreviouslySkippedEntriesEmittedOnNextIterationMinutesPrecision() throws Exception {
+ testPreviouslySkippedEntriesEmmitedOnNextIteration(TimeUnit.MINUTES);
+ }
+
+ private void testOnlyNewEntriesEmitted(final TimeUnit targetPrecision) throws InterruptedException {
+
+ final long initialTimestamp = getCurrentTimestampMillis(targetPrecision);
+
+ setTargetSystemTimestampPrecision(targetPrecision);
+
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ proc.addEntity("name", "id", initialTimestamp);
+ proc.addEntity("name", "id2", initialTimestamp);
+ runner.run();
+
+ // First run, the above listed entries would be skipped to avoid write synchronization issues
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
+ Thread.sleep(getSleepMillis(targetPrecision));
+
+ // Running again, our two previously seen files are now cleared to be released
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
+ runner.clearTransferState();
+
+ // Verify no new old files show up
+ proc.addEntity("name", "id2", initialTimestamp);
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // An entry that is older than already processed entry should not be listed.
+ proc.addEntity("name", "id3", initialTimestamp - targetPrecision.toMillis(1));
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // If an entry whose timestamp is the same with the last processed timestamp should not be listed.
+ proc.addEntity("name", "id2", initialTimestamp);
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Now a new file beyond the current time enters
+ proc.addEntity("name", "id2", initialTimestamp + targetPrecision.toMillis(1));
+
+ // It should show up
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 1);
+ runner.clearTransferState();
+ }
+
+ private void setTargetSystemTimestampPrecision(TimeUnit targetPrecision) {
+ switch (targetPrecision) {
+ case MINUTES:
+ runner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, PRECISION_MINUTES);
+ break;
+ case SECONDS:
+ runner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, PRECISION_SECONDS);
+ break;
+ case MILLISECONDS:
+ runner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, PRECISION_MILLIS);
+ break;
+ }
+ }
+
+ @Test
+ public void testOnlyNewEntriesEmittedMillisPrecision() throws Exception {
+ testOnlyNewEntriesEmitted(TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ public void testOnlyNewEntriesEmittedSecondPrecision() throws Exception {
+ testOnlyNewEntriesEmitted(TimeUnit.SECONDS);
+ }
+
+ @Test
+ public void testOnlyNewEntriesEmittedMinutesPrecision() throws Exception {
+ testOnlyNewEntriesEmitted(TimeUnit.MINUTES);
+ }
+
+ @Test
+ public void testHandleRestartWithEntriesAlreadyTransferredAndNoneNew() throws Exception {
+
+ final long initialTimestamp = System.currentTimeMillis();
+
+ proc.addEntity("name", "id", initialTimestamp);
+ proc.addEntity("name", "id2", initialTimestamp);
+
+ // Emulate having state but not having had the processor run such as in a restart
+ final Map<String, String> preexistingState = new HashMap<>();
+ preexistingState.put(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY, Long.toString(initialTimestamp));
+ preexistingState.put(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY, Long.toString(initialTimestamp));
+ preexistingState.put(AbstractListProcessor.IDENTIFIER_PREFIX + ".0", "id");
+ preexistingState.put(AbstractListProcessor.IDENTIFIER_PREFIX + ".1", "id2");
+ runner.getStateManager().setState(preexistingState, Scope.CLUSTER);
+
+ // run for the first time
+ runner.run();
+
+ // First run, the above listed entries would be skipped
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
+ Thread.sleep(DEFAULT_SLEEP_MILLIS);
+
+ // Running again, these files should be eligible for transfer and again skipped
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Verify no new old files show up
+ proc.addEntity("name", "id2", initialTimestamp);
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ proc.addEntity("name", "id3", initialTimestamp - 1);
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ proc.addEntity("name", "id2", initialTimestamp);
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Now a new file beyond the current time enters
+ proc.addEntity("name", "id2", initialTimestamp + 1);
+
+ // It should now show up
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 1);
+ runner.clearTransferState();
+ }
+
+ @Test
+ public void testStateStoredInClusterStateManagement() throws Exception {
+
+ final DistributedCache cache = new DistributedCache();
+ runner.addControllerService("cache", cache);
+ runner.enableControllerService(cache);
+ runner.setProperty(AbstractListProcessor.DISTRIBUTED_CACHE_SERVICE, "cache");
+
+ final long initialTimestamp = System.currentTimeMillis();
+
+ proc.addEntity("name", "id", initialTimestamp);
+ runner.run();
+
+ final Map<String, String> expectedState = new HashMap<>();
+ // Ensure only timestamp is migrated
+ expectedState.put(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY, String.valueOf(initialTimestamp));
+ expectedState.put(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY, "0");
+ runner.getStateManager().assertStateEquals(expectedState, Scope.CLUSTER);
+
+ Thread.sleep(DEFAULT_SLEEP_MILLIS);
+
+ runner.run();
+ // Ensure only timestamp is migrated
+ expectedState.put(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY, String.valueOf(initialTimestamp));
+ expectedState.put(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY, String.valueOf(initialTimestamp));
+ expectedState.put(AbstractListProcessor.IDENTIFIER_PREFIX + ".0", "id");
+ runner.getStateManager().assertStateEquals(expectedState, Scope.CLUSTER);
+ }
+
+ @Test
+ public void testResumeListingAfterClearingState() throws Exception {
+
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+
+ final long initialEventTimestamp = System.currentTimeMillis();
+ proc.addEntity("name", "id", initialEventTimestamp);
+ proc.addEntity("name", "id2", initialEventTimestamp);
+
+ // Add entities but these should not be transferred as they are the latest values
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+
+ // after providing a pause in listings, the files should now transfer
+ Thread.sleep(DEFAULT_SLEEP_MILLIS);
+
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
+ runner.clearTransferState();
+
+ // Verify entities are not transferred again for the given state
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ // Clear state for this processor, eradicating timestamp
+ runner.getStateManager().clear(Scope.CLUSTER);
+ Assert.assertEquals("State is not empty for this component after clearing", 0, runner.getStateManager().getState(Scope.CLUSTER).toMap().size());
+
+ // Ensure the original files are now transferred again.
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
+ runner.clearTransferState();
+ }
+
+ @Test
+ public void testOnlyNewStateStored() throws Exception {
+
+ runner.run();
+
+ final long initialTimestamp = System.currentTimeMillis();
+
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ proc.addEntity("name", "id", initialTimestamp);
+ proc.addEntity("name", "id2", initialTimestamp);
+
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
+ runner.clearTransferState();
+
+ Thread.sleep(DEFAULT_SLEEP_MILLIS);
+
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
+ runner.clearTransferState();
+
+ final StateMap stateMap = runner.getStateManager().getState(Scope.CLUSTER);
+ assertEquals(2, stateMap.getVersion());
+
+ final Map<String, String> map = stateMap.toMap();
+ // Ensure timestamp and identifiers are migrated
+ assertEquals(4, map.size());
+ assertEquals(Long.toString(initialTimestamp), map.get(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY));
+ assertEquals(Long.toString(initialTimestamp), map.get(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY));
+ assertEquals("id", map.get(AbstractListProcessor.IDENTIFIER_PREFIX + ".0"));
+ assertEquals("id2", map.get(AbstractListProcessor.IDENTIFIER_PREFIX + ".1"));
+
+ proc.addEntity("new name", "new id", initialTimestamp + 1);
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 1);
+ runner.clearTransferState();
+
+ StateMap updatedStateMap = runner.getStateManager().getState(Scope.CLUSTER);
+ assertEquals(3, updatedStateMap.getVersion());
+
+ assertEquals(3, updatedStateMap.toMap().size());
+ assertEquals(Long.toString(initialTimestamp + 1), updatedStateMap.get(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY));
+ // Processed timestamp is now caught up
+ assertEquals(Long.toString(initialTimestamp + 1), updatedStateMap.get(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY));
+ assertEquals("new id", updatedStateMap.get(AbstractListProcessor.IDENTIFIER_PREFIX + ".0"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/TestAbstractListProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/TestAbstractListProcessor.java b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/TestAbstractListProcessor.java
index 4a376e2..f5eae46 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/TestAbstractListProcessor.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/test/java/org/apache/nifi/processor/util/list/TestAbstractListProcessor.java
@@ -19,7 +19,6 @@ package org.apache.nifi.processor.util.list;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.Scope;
-import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.distributed.cache.client.Deserializer;
import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient;
@@ -33,7 +32,6 @@ import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Assert;
import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
@@ -54,10 +52,6 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
-import static org.apache.nifi.processor.util.list.AbstractListProcessor.PRECISION_MILLIS;
-import static org.apache.nifi.processor.util.list.AbstractListProcessor.PRECISION_MINUTES;
-import static org.apache.nifi.processor.util.list.AbstractListProcessor.PRECISION_SECONDS;
-import static org.apache.nifi.processor.util.list.AbstractListProcessor.TARGET_SYSTEM_TIMESTAMP_PRECISION;
import static org.junit.Assert.assertEquals;
public class TestAbstractListProcessor {
@@ -101,257 +95,6 @@ public class TestAbstractListProcessor {
@Rule
public final TemporaryFolder testFolder = new TemporaryFolder();
- /**
- * <p>Ensures that files are listed when those are old enough:
- * <li>Files with last modified timestamp those are old enough to determine that those are completely written
- * and no further files are expected to be added with the same timestamp.</li>
- * <li>This behavior is expected when a processor is scheduled less frequently, such as hourly or daily.</li>
- * </p>
- */
- @Test
- public void testAllExistingEntriesEmittedOnFirstIteration() throws Exception {
- final long oldTimestamp = System.currentTimeMillis() - getSleepMillis(TimeUnit.MILLISECONDS);
-
- // These entries have existed before the processor runs at the first time.
- proc.addEntity("name", "id", oldTimestamp);
- proc.addEntity("name", "id2", oldTimestamp);
-
- // First run, the above listed entries should be emitted since it has existed.
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
- runner.clearTransferState();
-
- // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
- Thread.sleep(DEFAULT_SLEEP_MILLIS);
-
- // Run again without introducing any new entries
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- }
-
- private void testPreviouslySkippedEntriesEmmitedOnNextIteration(final TimeUnit targetPrecision) throws InterruptedException {
- runner.run();
-
- final long initialTimestamp = getCurrentTimestampMillis(targetPrecision);
-
- setTargetSystemTimestampPrecision(targetPrecision);
-
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- proc.addEntity("name", "id", initialTimestamp);
- proc.addEntity("name", "id2", initialTimestamp);
- runner.run();
-
- // First run, the above listed entries would be skipped to avoid write synchronization issues
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
- Thread.sleep(getSleepMillis(targetPrecision));
-
- // Run again without introducing any new entries
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
- }
-
- /**
- * <p>Ensures that newly created files should wait to confirm there is no more files created with the same timestamp:
- * <li>If files have the latest modified timestamp at an iteration, then those should be postponed to be listed</li>
- * <li>If those files still are the latest files at the next iteration, then those should be listed</li>
- * </p>
- */
- @Test
- public void testPreviouslySkippedEntriesEmittedOnNextIterationMilliPrecision() throws Exception {
- testPreviouslySkippedEntriesEmmitedOnNextIteration(TimeUnit.MILLISECONDS);
- }
-
- /**
- * Same as {@link #testPreviouslySkippedEntriesEmittedOnNextIterationMilliPrecision()} but simulates that the target
- * filesystem only provide timestamp precision in Seconds.
- */
- @Test
- public void testPreviouslySkippedEntriesEmittedOnNextIterationSecondPrecision() throws Exception {
- testPreviouslySkippedEntriesEmmitedOnNextIteration(TimeUnit.SECONDS);
- }
-
- /**
- * Same as {@link #testPreviouslySkippedEntriesEmittedOnNextIterationMilliPrecision()} but simulates that the target
- * filesystem only provide timestamp precision in Minutes.
- * This test is ignored because it needs to wait two minutes. Not good for automated unit testing, but still valuable when executed manually.
- */
- @Ignore
- @Test
- public void testPreviouslySkippedEntriesEmittedOnNextIterationMinutesPrecision() throws Exception {
- testPreviouslySkippedEntriesEmmitedOnNextIteration(TimeUnit.MINUTES);
- }
-
- private void testOnlyNewEntriesEmitted(final TimeUnit targetPrecision) throws InterruptedException {
-
- final long initialTimestamp = getCurrentTimestampMillis(targetPrecision);
-
- setTargetSystemTimestampPrecision(targetPrecision);
-
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- proc.addEntity("name", "id", initialTimestamp);
- proc.addEntity("name", "id2", initialTimestamp);
- runner.run();
-
- // First run, the above listed entries would be skipped to avoid write synchronization issues
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
- Thread.sleep(getSleepMillis(targetPrecision));
-
- // Running again, our two previously seen files are now cleared to be released
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
- runner.clearTransferState();
-
- // Verify no new old files show up
- proc.addEntity("name", "id2", initialTimestamp);
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // An entry that is older than already processed entry should not be listed.
- proc.addEntity("name", "id3", initialTimestamp - targetPrecision.toMillis(1));
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // If an entry whose timestamp is the same with the last processed timestamp should not be listed.
- proc.addEntity("name", "id2", initialTimestamp);
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Now a new file beyond the current time enters
- proc.addEntity("name", "id2", initialTimestamp + targetPrecision.toMillis(1));
-
- // It should show up
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 1);
- runner.clearTransferState();
- }
-
- private void setTargetSystemTimestampPrecision(TimeUnit targetPrecision) {
- switch (targetPrecision) {
- case MINUTES:
- runner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, PRECISION_MINUTES);
- break;
- case SECONDS:
- runner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, PRECISION_SECONDS);
- break;
- case MILLISECONDS:
- runner.setProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION, PRECISION_MILLIS);
- break;
- }
- }
-
- @Test
- public void testOnlyNewEntriesEmittedMillisPrecision() throws Exception {
- testOnlyNewEntriesEmitted(TimeUnit.MILLISECONDS);
- }
-
- @Test
- public void testOnlyNewEntriesEmittedSecondPrecision() throws Exception {
- testOnlyNewEntriesEmitted(TimeUnit.SECONDS);
- }
-
- /**
- * This test is ignored because it needs to wait two minutes. Not good for automated unit testing, but still valuable when executed manually.
- */
- @Ignore
- @Test
- public void testOnlyNewEntriesEmittedMinutesPrecision() throws Exception {
- testOnlyNewEntriesEmitted(TimeUnit.MINUTES);
- }
-
- @Test
- public void testHandleRestartWithEntriesAlreadyTransferredAndNoneNew() throws Exception {
-
- final long initialTimestamp = System.currentTimeMillis();
-
- proc.addEntity("name", "id", initialTimestamp);
- proc.addEntity("name", "id2", initialTimestamp);
-
- // Emulate having state but not having had the processor run such as in a restart
- final Map<String, String> preexistingState = new HashMap<>();
- preexistingState.put(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY, Long.toString(initialTimestamp));
- preexistingState.put(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY, Long.toString(initialTimestamp));
- preexistingState.put(AbstractListProcessor.IDENTIFIER_PREFIX + ".0", "id");
- preexistingState.put(AbstractListProcessor.IDENTIFIER_PREFIX + ".1", "id2");
- runner.getStateManager().setState(preexistingState, Scope.CLUSTER);
-
- // run for the first time
- runner.run();
-
- // First run, the above listed entries would be skipped
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Ensure we have covered the necessary lag period to avoid issues where the processor was immediately scheduled to run again
- Thread.sleep(DEFAULT_SLEEP_MILLIS);
-
- // Running again, these files should be eligible for transfer and again skipped
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Verify no new old files show up
- proc.addEntity("name", "id2", initialTimestamp);
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- proc.addEntity("name", "id3", initialTimestamp - 1);
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- proc.addEntity("name", "id2", initialTimestamp);
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Now a new file beyond the current time enters
- proc.addEntity("name", "id2", initialTimestamp + 1);
-
- // It should now show up
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 1);
- runner.clearTransferState();
- }
-
- @Test
- public void testStateStoredInClusterStateManagement() throws Exception {
-
- final DistributedCache cache = new DistributedCache();
- runner.addControllerService("cache", cache);
- runner.enableControllerService(cache);
- runner.setProperty(AbstractListProcessor.DISTRIBUTED_CACHE_SERVICE, "cache");
-
- final long initialTimestamp = System.currentTimeMillis();
-
- proc.addEntity("name", "id", initialTimestamp);
- runner.run();
-
- final Map<String, String> expectedState = new HashMap<>();
- // Ensure only timestamp is migrated
- expectedState.put(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY, String.valueOf(initialTimestamp));
- expectedState.put(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY, "0");
- runner.getStateManager().assertStateEquals(expectedState, Scope.CLUSTER);
-
- Thread.sleep(DEFAULT_SLEEP_MILLIS);
-
- runner.run();
- // Ensure only timestamp is migrated
- expectedState.put(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY, String.valueOf(initialTimestamp));
- expectedState.put(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY, String.valueOf(initialTimestamp));
- expectedState.put(AbstractListProcessor.IDENTIFIER_PREFIX + ".0", "id");
- runner.getStateManager().assertStateEquals(expectedState, Scope.CLUSTER);
- }
-
@Test
public void testStateMigratedFromCacheService() throws InitializationException {
@@ -416,41 +159,6 @@ public class TestAbstractListProcessor {
}
@Test
- public void testResumeListingAfterClearingState() throws Exception {
-
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
-
- final long initialEventTimestamp = System.currentTimeMillis();
- proc.addEntity("name", "id", initialEventTimestamp);
- proc.addEntity("name", "id2", initialEventTimestamp);
-
- // Add entities but these should not be transferred as they are the latest values
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
-
- // after providing a pause in listings, the files should now transfer
- Thread.sleep(DEFAULT_SLEEP_MILLIS);
-
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
- runner.clearTransferState();
-
- // Verify entities are not transferred again for the given state
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- // Clear state for this processor, eradicating timestamp
- runner.getStateManager().clear(Scope.CLUSTER);
- Assert.assertEquals("State is not empty for this component after clearing", 0, runner.getStateManager().getState(Scope.CLUSTER).toMap().size());
-
- // Ensure the original files are now transferred again.
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
- runner.clearTransferState();
- }
-
- @Test
public void testFetchOnStart() throws InitializationException {
final DistributedCache cache = new DistributedCache();
@@ -463,55 +171,7 @@ public class TestAbstractListProcessor {
assertEquals(1, cache.fetchCount);
}
- @Test
- public void testOnlyNewStateStored() throws Exception {
-
- runner.run();
-
- final long initialTimestamp = System.currentTimeMillis();
-
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- proc.addEntity("name", "id", initialTimestamp);
- proc.addEntity("name", "id2", initialTimestamp);
-
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 0);
- runner.clearTransferState();
-
- Thread.sleep(DEFAULT_SLEEP_MILLIS);
-
- runner.run();
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 2);
- runner.clearTransferState();
-
- final StateMap stateMap = runner.getStateManager().getState(Scope.CLUSTER);
- assertEquals(2, stateMap.getVersion());
-
- final Map<String, String> map = stateMap.toMap();
- // Ensure timestamp and identifiers are migrated
- assertEquals(4, map.size());
- assertEquals(Long.toString(initialTimestamp), map.get(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY));
- assertEquals(Long.toString(initialTimestamp), map.get(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY));
- assertEquals("id", map.get(AbstractListProcessor.IDENTIFIER_PREFIX + ".0"));
- assertEquals("id2", map.get(AbstractListProcessor.IDENTIFIER_PREFIX + ".1"));
-
- proc.addEntity("new name", "new id", initialTimestamp + 1);
- runner.run();
-
- runner.assertAllFlowFilesTransferred(ConcreteListProcessor.REL_SUCCESS, 1);
- runner.clearTransferState();
-
- StateMap updatedStateMap = runner.getStateManager().getState(Scope.CLUSTER);
- assertEquals(3, updatedStateMap.getVersion());
-
- assertEquals(3, updatedStateMap.toMap().size());
- assertEquals(Long.toString(initialTimestamp + 1), updatedStateMap.get(AbstractListProcessor.LATEST_LISTED_ENTRY_TIMESTAMP_KEY));
- // Processed timestamp is now caught up
- assertEquals(Long.toString(initialTimestamp + 1), updatedStateMap.get(AbstractListProcessor.LAST_PROCESSED_LATEST_ENTRY_TIMESTAMP_KEY));
- assertEquals("new id", updatedStateMap.get(AbstractListProcessor.IDENTIFIER_PREFIX + ".0"));
- }
-
- private static class DistributedCache extends AbstractControllerService implements DistributedMapCacheClient {
+ static class DistributedCache extends AbstractControllerService implements DistributedMapCacheClient {
private final Map<Object, Object> stored = new HashMap<>();
private int fetchCount = 0;
@@ -569,13 +229,12 @@ public class TestAbstractListProcessor {
}
}
+ static class ConcreteListProcessor extends AbstractListProcessor<ListableEntity> {
+ final List<ListableEntity> entities = new ArrayList<>();
- private static class ConcreteListProcessor extends AbstractListProcessor<ListableEntity> {
- private final List<ListableEntity> entities = new ArrayList<>();
-
- public final String persistenceFilename = "ListProcessor-local-state-" + UUID.randomUUID().toString() + ".json";
- public String persistenceFolder = "target/";
- public File persistenceFile = new File(persistenceFolder + persistenceFilename);
+ final String persistenceFilename = "ListProcessor-local-state-" + UUID.randomUUID().toString() + ".json";
+ String persistenceFolder = "target/";
+ File persistenceFile = new File(persistenceFolder + persistenceFilename);
@Override
public File getPersistenceFile() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
index bd93e3b..3496ea0 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
@@ -18,17 +18,10 @@ package org.apache.nifi.processors.kafka.pubsub;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Before;
@@ -100,97 +93,6 @@ public class ConsumeKafkaTest {
}
@Test
- public void validateGetAllMessages() throws Exception {
- String groupName = "validateGetAllMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka_0_10 proc = new ConsumeKafka_0_10() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_0_10.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka_0_10.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_0_10.AUTO_OFFSET_RESET, ConsumeKafka_0_10.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetAllMessagesPattern() throws Exception {
- String groupName = "validateGetAllMessagesPattern";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka_0_10 proc = new ConsumeKafka_0_10() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_0_10.TOPICS, "(fo.*)|(ba)");
- runner.setProperty(ConsumeKafka_0_10.TOPIC_TYPE, "pattern");
- runner.setProperty(ConsumeKafka_0_10.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_0_10.AUTO_OFFSET_RESET, ConsumeKafka_0_10.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetErrorMessages() throws Exception {
- String groupName = "validateGetErrorMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(true, false);
- when(mockLease.commit()).thenReturn(Boolean.FALSE);
-
- ConsumeKafka_0_10 proc = new ConsumeKafka_0_10() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_0_10.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka_0_10.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_0_10.AUTO_OFFSET_RESET, ConsumeKafka_0_10.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(2)).continuePolling();
- verify(mockLease, times(1)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
public void testJaasConfiguration() throws Exception {
ConsumeKafka_0_10 consumeKafka = new ConsumeKafka_0_10();
TestRunner runner = TestRunners.newTestRunner(consumeKafka);
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
new file mode 100644
index 0000000..b0c6eba
--- /dev/null
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-10-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
@@ -0,0 +1,135 @@
+/*
+ * 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.nifi.processors.kafka.pubsub;
+
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ITConsumeKafka {
+
+ ConsumerLease mockLease = null;
+ ConsumerPool mockConsumerPool = null;
+
+ @Before
+ public void setup() {
+ mockLease = mock(ConsumerLease.class);
+ mockConsumerPool = mock(ConsumerPool.class);
+ }
+
+ @Test
+ public void validateGetAllMessages() throws Exception {
+ String groupName = "validateGetAllMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka_0_10 proc = new ConsumeKafka_0_10() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_0_10.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka_0_10.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_0_10.AUTO_OFFSET_RESET, ConsumeKafka_0_10.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetAllMessagesPattern() throws Exception {
+ String groupName = "validateGetAllMessagesPattern";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka_0_10 proc = new ConsumeKafka_0_10() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_0_10.TOPICS, "(fo.*)|(ba)");
+ runner.setProperty(ConsumeKafka_0_10.TOPIC_TYPE, "pattern");
+ runner.setProperty(ConsumeKafka_0_10.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_0_10.AUTO_OFFSET_RESET, ConsumeKafka_0_10.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetErrorMessages() throws Exception {
+ String groupName = "validateGetErrorMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(true, false);
+ when(mockLease.commit()).thenReturn(Boolean.FALSE);
+
+ ConsumeKafka_0_10 proc = new ConsumeKafka_0_10() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_0_10.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka_0_10.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_0_10.AUTO_OFFSET_RESET, ConsumeKafka_0_10.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(2)).continuePolling();
+ verify(mockLease, times(1)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
index e460879..b1edd1f 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
@@ -18,17 +18,10 @@ package org.apache.nifi.processors.kafka.pubsub;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Before;
@@ -100,97 +93,6 @@ public class ConsumeKafkaTest {
}
@Test
- public void validateGetAllMessages() throws Exception {
- String groupName = "validateGetAllMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka_0_11 proc = new ConsumeKafka_0_11() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_0_11.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka_0_11.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_0_11.AUTO_OFFSET_RESET, ConsumeKafka_0_11.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetAllMessagesPattern() throws Exception {
- String groupName = "validateGetAllMessagesPattern";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka_0_11 proc = new ConsumeKafka_0_11() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_0_11.TOPICS, "(fo.*)|(ba)");
- runner.setProperty(ConsumeKafka_0_11.TOPIC_TYPE, "pattern");
- runner.setProperty(ConsumeKafka_0_11.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_0_11.AUTO_OFFSET_RESET, ConsumeKafka_0_11.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetErrorMessages() throws Exception {
- String groupName = "validateGetErrorMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(true, false);
- when(mockLease.commit()).thenReturn(Boolean.FALSE);
-
- ConsumeKafka_0_11 proc = new ConsumeKafka_0_11() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_0_11.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka_0_11.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_0_11.AUTO_OFFSET_RESET, ConsumeKafka_0_11.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(2)).continuePolling();
- verify(mockLease, times(1)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
public void testJaasConfiguration() throws Exception {
ConsumeKafka_0_11 consumeKafka = new ConsumeKafka_0_11();
TestRunner runner = TestRunners.newTestRunner(consumeKafka);
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
new file mode 100644
index 0000000..9a3b3d9
--- /dev/null
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-11-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
@@ -0,0 +1,135 @@
+/*
+ * 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.nifi.processors.kafka.pubsub;
+
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ITConsumeKafka {
+
+ ConsumerLease mockLease = null;
+ ConsumerPool mockConsumerPool = null;
+
+ @Before
+ public void setup() {
+ mockLease = mock(ConsumerLease.class);
+ mockConsumerPool = mock(ConsumerPool.class);
+ }
+
+ @Test
+ public void validateGetAllMessages() throws Exception {
+ String groupName = "validateGetAllMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka_0_11 proc = new ConsumeKafka_0_11() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_0_11.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka_0_11.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_0_11.AUTO_OFFSET_RESET, ConsumeKafka_0_11.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetAllMessagesPattern() throws Exception {
+ String groupName = "validateGetAllMessagesPattern";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka_0_11 proc = new ConsumeKafka_0_11() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_0_11.TOPICS, "(fo.*)|(ba)");
+ runner.setProperty(ConsumeKafka_0_11.TOPIC_TYPE, "pattern");
+ runner.setProperty(ConsumeKafka_0_11.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_0_11.AUTO_OFFSET_RESET, ConsumeKafka_0_11.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetErrorMessages() throws Exception {
+ String groupName = "validateGetErrorMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(true, false);
+ when(mockLease.commit()).thenReturn(Boolean.FALSE);
+
+ ConsumeKafka_0_11 proc = new ConsumeKafka_0_11() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka_0_11.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka_0_11.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka_0_11.AUTO_OFFSET_RESET, ConsumeKafka_0_11.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
+ verify(mockLease, times(2)).continuePolling();
+ verify(mockLease, times(1)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
index c4b0140..9062e1e 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
@@ -20,24 +20,13 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.nifi.components.PropertyValue;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Test;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.junit.Assume.assumeFalse;
import org.junit.Before;
-import static org.mockito.Matchers.anyObject;
-import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
public class ConsumeKafkaTest {
@@ -105,103 +94,4 @@ public class ConsumeKafkaTest {
assertTrue(e.getMessage().contains("must contain at least one character that is not white space"));
}
}
-
- @Test
- public void validateGetAllMessages() throws Exception {
- String groupName = "validateGetAllMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka proc = new ConsumeKafka() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET, ConsumeKafka.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetErrorMessages() throws Exception {
- String groupName = "validateGetErrorMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(true, false);
- when(mockLease.commit()).thenReturn(Boolean.FALSE);
-
- ConsumeKafka proc = new ConsumeKafka() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET, ConsumeKafka.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject());
- verify(mockLease, times(2)).continuePolling();
- verify(mockLease, times(1)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- private boolean isWindowsEnvironment() {
- return System.getProperty("os.name").toLowerCase().startsWith("windows");
- }
-
- @Test
- public void validateConsumerRetainer() throws Exception {
- assumeFalse(isWindowsEnvironment());//skip if on windows
- final ConsumerPool consumerPool = mock(ConsumerPool.class);
-
- final ConsumeKafka processor = new ConsumeKafka() {
- @Override
- protected ConsumerPool createConsumerPool(ProcessContext context, ComponentLog log) {
- return consumerPool;
- }
- };
-
- final ComponentLog logger = mock(ComponentLog.class);
- final ProcessorInitializationContext initializationContext = mock(ProcessorInitializationContext.class);
- when(initializationContext.getLogger()).thenReturn(logger);
- processor.initialize(initializationContext);
-
- final ProcessContext processContext = mock(ProcessContext.class);
- final PropertyValue heartbeatInternalMsConfig = mock(PropertyValue.class);
- when(heartbeatInternalMsConfig.isSet()).thenReturn(true);
- when(heartbeatInternalMsConfig.asInteger()).thenReturn(100);
- when(processContext.getProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG)).thenReturn(heartbeatInternalMsConfig);
- processor.onScheduled(processContext);
-
- // retainConsumers should be called at least 1 time if it passed longer than heartbeat interval milliseconds.
- Thread.sleep(200);
- verify(consumerPool, atLeast(1)).retainConsumers();
-
- processor.stopConnectionRetainer();
-
- // After stopping connection retainer, it shouldn't interact with consumerPool.
- Thread.sleep(200);
- verifyNoMoreInteractions(consumerPool);
- }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
new file mode 100644
index 0000000..084280a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ITConsumeKafka.java
@@ -0,0 +1,150 @@
+/*
+ * 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.nifi.processors.kafka.pubsub;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+import org.apache.nifi.components.PropertyValue;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+import static org.junit.Assume.assumeFalse;
+import org.junit.Before;
+import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
+
+public class ITConsumeKafka {
+
+ Consumer<byte[], byte[]> mockConsumer = null;
+ ConsumerLease mockLease = null;
+ ConsumerPool mockConsumerPool = null;
+
+ @Before
+ public void setup() {
+ mockConsumer = mock(Consumer.class);
+ mockLease = mock(ConsumerLease.class);
+ mockConsumerPool = mock(ConsumerPool.class);
+ }
+
+ @Test
+ public void validateGetAllMessages() throws Exception {
+ String groupName = "validateGetAllMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
+ when(mockLease.commit()).thenReturn(Boolean.TRUE);
+
+ ConsumeKafka proc = new ConsumeKafka() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET, ConsumeKafka.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject());
+ verify(mockLease, times(3)).continuePolling();
+ verify(mockLease, times(2)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ @Test
+ public void validateGetErrorMessages() throws Exception {
+ String groupName = "validateGetErrorMessages";
+
+ when(mockConsumerPool.obtainConsumer(anyObject())).thenReturn(mockLease);
+ when(mockLease.continuePolling()).thenReturn(true, false);
+ when(mockLease.commit()).thenReturn(Boolean.FALSE);
+
+ ConsumeKafka proc = new ConsumeKafka() {
+ @Override
+ protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
+ return mockConsumerPool;
+ }
+ };
+ final TestRunner runner = TestRunners.newTestRunner(proc);
+ runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
+ runner.setProperty(ConsumeKafka.TOPICS, "foo,bar");
+ runner.setProperty(ConsumeKafka.GROUP_ID, groupName);
+ runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET, ConsumeKafka.OFFSET_EARLIEST);
+ runner.run(1, false);
+
+ verify(mockConsumerPool, times(1)).obtainConsumer(anyObject());
+ verify(mockLease, times(2)).continuePolling();
+ verify(mockLease, times(1)).poll();
+ verify(mockLease, times(1)).commit();
+ verify(mockLease, times(1)).close();
+ verifyNoMoreInteractions(mockConsumerPool);
+ verifyNoMoreInteractions(mockLease);
+ }
+
+ private boolean isWindowsEnvironment() {
+ return System.getProperty("os.name").toLowerCase().startsWith("windows");
+ }
+
+ @Test
+ public void validateConsumerRetainer() throws Exception {
+ assumeFalse(isWindowsEnvironment());//skip if on windows
+ final ConsumerPool consumerPool = mock(ConsumerPool.class);
+
+ final ConsumeKafka processor = new ConsumeKafka() {
+ @Override
+ protected ConsumerPool createConsumerPool(ProcessContext context, ComponentLog log) {
+ return consumerPool;
+ }
+ };
+
+ final ComponentLog logger = mock(ComponentLog.class);
+ final ProcessorInitializationContext initializationContext = mock(ProcessorInitializationContext.class);
+ when(initializationContext.getLogger()).thenReturn(logger);
+ processor.initialize(initializationContext);
+
+ final ProcessContext processContext = mock(ProcessContext.class);
+ final PropertyValue heartbeatInternalMsConfig = mock(PropertyValue.class);
+ when(heartbeatInternalMsConfig.isSet()).thenReturn(true);
+ when(heartbeatInternalMsConfig.asInteger()).thenReturn(100);
+ when(processContext.getProperty(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG)).thenReturn(heartbeatInternalMsConfig);
+ processor.onScheduled(processContext);
+
+ // retainConsumers should be called at least 1 time if it passed longer than heartbeat interval milliseconds.
+ Thread.sleep(200);
+ verify(consumerPool, atLeast(1)).retainConsumers();
+
+ processor.stopConnectionRetainer();
+
+ // After stopping connection retainer, it shouldn't interact with consumerPool.
+ Thread.sleep(200);
+ verifyNoMoreInteractions(consumerPool);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/cdc1facf/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
index 7b5a8fc..10ac398 100644
--- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
+++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-1-0-processors/src/test/java/org/apache/nifi/processors/kafka/pubsub/ConsumeKafkaTest.java
@@ -18,17 +18,10 @@ package org.apache.nifi.processors.kafka.pubsub;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.mockito.Matchers.anyObject;
import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.verifyNoMoreInteractions;
-import static org.mockito.Mockito.when;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Before;
@@ -100,97 +93,6 @@ public class ConsumeKafkaTest {
}
@Test
- public void validateGetAllMessages() throws Exception {
- String groupName = "validateGetAllMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka_1_0 proc = new ConsumeKafka_1_0() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_1_0.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka_1_0.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_1_0.AUTO_OFFSET_RESET, ConsumeKafka_1_0.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetAllMessagesPattern() throws Exception {
- String groupName = "validateGetAllMessagesPattern";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE);
- when(mockLease.commit()).thenReturn(Boolean.TRUE);
-
- ConsumeKafka_1_0 proc = new ConsumeKafka_1_0() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_1_0.TOPICS, "(fo.*)|(ba)");
- runner.setProperty(ConsumeKafka_1_0.TOPIC_TYPE, "pattern");
- runner.setProperty(ConsumeKafka_1_0.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_1_0.AUTO_OFFSET_RESET, ConsumeKafka_1_0.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(3)).continuePolling();
- verify(mockLease, times(2)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
- public void validateGetErrorMessages() throws Exception {
- String groupName = "validateGetErrorMessages";
-
- when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease);
- when(mockLease.continuePolling()).thenReturn(true, false);
- when(mockLease.commit()).thenReturn(Boolean.FALSE);
-
- ConsumeKafka_1_0 proc = new ConsumeKafka_1_0() {
- @Override
- protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) {
- return mockConsumerPool;
- }
- };
- final TestRunner runner = TestRunners.newTestRunner(proc);
- runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234");
- runner.setProperty(ConsumeKafka_1_0.TOPICS, "foo,bar");
- runner.setProperty(ConsumeKafka_1_0.GROUP_ID, groupName);
- runner.setProperty(ConsumeKafka_1_0.AUTO_OFFSET_RESET, ConsumeKafka_1_0.OFFSET_EARLIEST);
- runner.run(1, false);
-
- verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), anyObject());
- verify(mockLease, times(2)).continuePolling();
- verify(mockLease, times(1)).poll();
- verify(mockLease, times(1)).commit();
- verify(mockLease, times(1)).close();
- verifyNoMoreInteractions(mockConsumerPool);
- verifyNoMoreInteractions(mockLease);
- }
-
- @Test
public void testJaasConfiguration() throws Exception {
ConsumeKafka_1_0 consumeKafka = new ConsumeKafka_1_0();
TestRunner runner = TestRunners.newTestRunner(consumeKafka);