From d15787bcd99afca40c5346a028319a1f8bb16b87 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 14 May 2018 10:03:46 -0400 Subject: [PATCH] Remove SIK Extension code Fixes https://github.com/spring-projects/spring-integration-extensions/issues/192 --- .../support/ConsumerConfigurationTests.java | 610 ------------------ 1 file changed, 610 deletions(-) delete mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java deleted file mode 100644 index 86692fc..0000000 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/support/ConsumerConfigurationTests.java +++ /dev/null @@ -1,610 +0,0 @@ -/* - * Copyright 2002-2013 the original author or authors. - * - * Licensed 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.springframework.integration.kafka.support; - -import static org.junit.Assert.assertNull; -import static org.mockito.Mockito.*; - -import java.util.*; - -import kafka.consumer.ConsumerIterator; -import kafka.consumer.KafkaStream; -import kafka.javaapi.consumer.ConsumerConnector; -import kafka.message.MessageAndMetadata; -import kafka.serializer.Decoder; - -import org.junit.Assert; -import org.junit.Test; - -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; - -/** - * @author Soby Chacko - * @author Rajasekar Elango - * @since 0.5 - */ -public class ConsumerConfigurationTests { - - @Test - @SuppressWarnings("unchecked") - public void testReceiveMessageForSingleTopicFromSingleStream() { - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); - final ConsumerConnectionProvider consumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); - final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); - - Map topicStreamMap = new HashMap(); - topicStreamMap.put("topic1", 1); - when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); - - when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); - consumerConfiguration.setMaxMessages(1); - - final KafkaStream stream = mock(KafkaStream.class); - final List> streams = new ArrayList>(); - streams.add(stream); - final Map>> messageStreams = new HashMap>>(); - messageStreams.put("topic", streams); - - when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); - when(stream.iterator()).thenReturn(iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); - when(iterator.next()).thenReturn(messageAndMetadata); - when(messageAndMetadata.message()).thenReturn((V) "got message"); - when(messageAndMetadata.topic()).thenReturn("topic"); - when(messageAndMetadata.partition()).thenReturn(1); - - final Map>> messages = consumerConfiguration.receive(); - Assert.assertEquals(1, messages.size()); - Assert.assertEquals(1, messages.get("topic").size()); - Assert.assertEquals("got message", messages.get("topic").get(1).get(0)); - - verify(stream, times(1)).iterator(); - verify(iterator, times(1)).next(); - verify(messageAndMetadata, times(1)).message(); - verify(messageAndMetadata, times(1)).topic(); - } - - @Test - @SuppressWarnings("unchecked") - public void testReceiveMessageForSingleTopicFromMultipleStreams() { -<<<<<<< HEAD - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); -======= - final ConsumerMetadata consumerMetadata = Mockito.mock(ConsumerMetadata.class); ->>>>>>> INTEXT-40 - Add ZIP Transformer - final ConsumerConnectionProvider consumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); - - Map topicStreamMap = new HashMap(); - topicStreamMap.put("topic1", 1); - when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); - - final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); - - when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); - consumerConfiguration.setMaxMessages(3); - - final KafkaStream stream1 = mock(KafkaStream.class); - final KafkaStream stream2 = mock(KafkaStream.class); - final KafkaStream stream3 = mock(KafkaStream.class); - final List> streams = new ArrayList>(); - streams.add(stream1); - streams.add(stream2); - streams.add(stream3); - final Map>> messageStreams = new HashMap>>(); - messageStreams.put("topic", streams); - - when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); - final ConsumerIterator iterator1 = mock(ConsumerIterator.class); - final ConsumerIterator iterator2 = mock(ConsumerIterator.class); - final ConsumerIterator iterator3 = mock(ConsumerIterator.class); - - when(stream1.iterator()).thenReturn(iterator1); - when(stream2.iterator()).thenReturn(iterator2); - when(stream3.iterator()).thenReturn(iterator3); - final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class); - final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class); - final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class); - - when(iterator1.next()).thenReturn(messageAndMetadata1); - when(iterator2.next()).thenReturn(messageAndMetadata2); - when(iterator3.next()).thenReturn(messageAndMetadata3); - - when(messageAndMetadata1.message()).thenReturn((V)"got message"); - when(messageAndMetadata1.topic()).thenReturn("topic"); - when(messageAndMetadata1.partition()).thenReturn(1); - - when(messageAndMetadata2.message()).thenReturn((V)"got message"); - when(messageAndMetadata2.topic()).thenReturn("topic"); - when(messageAndMetadata2.partition()).thenReturn(2); - - when(messageAndMetadata3.message()).thenReturn((V)"got message"); - when(messageAndMetadata3.topic()).thenReturn("topic"); - when(messageAndMetadata3.partition()).thenReturn(3); - - final Map>> messages = consumerConfiguration.receive(); - Assert.assertEquals(messages.size(), 1); - int sum = 0; - - final Map> values = messages.get("topic"); - - for (final List l : values.values()) { - sum += l.size(); - } - - Assert.assertEquals(3, sum); - } - - @Test - @SuppressWarnings("unchecked") - public void testReceiveMessageForMultipleTopicsFromMultipleStreams() { -<<<<<<< HEAD - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); -======= - final ConsumerMetadata consumerMetadata = Mockito.mock(ConsumerMetadata.class); ->>>>>>> INTEXT-40 - Add ZIP Transformer - final ConsumerConnectionProvider consumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); - - Map topicStreamMap = new HashMap(); - topicStreamMap.put("topic1", 1); - when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); - - - final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); - - when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); - consumerConfiguration.setMaxMessages(9); - - final KafkaStream stream1 = mock(KafkaStream.class); - final KafkaStream stream2 = mock(KafkaStream.class); - final KafkaStream stream3 = mock(KafkaStream.class); - final List> streams = new ArrayList>(); - streams.add(stream1); - streams.add(stream2); - streams.add(stream3); - final Map>> messageStreams = new HashMap>>(); - messageStreams.put("topic1", streams); - messageStreams.put("topic2", streams); - messageStreams.put("topic3", streams); - - when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); - final ConsumerIterator iterator1 = mock(ConsumerIterator.class); - final ConsumerIterator iterator2 = mock(ConsumerIterator.class); - final ConsumerIterator iterator3 = mock(ConsumerIterator.class); - - when(stream1.iterator()).thenReturn(iterator1); - when(stream2.iterator()).thenReturn(iterator2); - when(stream3.iterator()).thenReturn(iterator3); - final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class); - final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class); - final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class); - - when(iterator1.next()).thenReturn(messageAndMetadata1); - when(iterator2.next()).thenReturn(messageAndMetadata2); - when(iterator3.next()).thenReturn(messageAndMetadata3); - - when(messageAndMetadata1.message()).thenReturn((V)"got message1"); - when(messageAndMetadata1.topic()).thenReturn("topic1"); - when(messageAndMetadata1.partition()).thenAnswer(getAnswer()); - - when(messageAndMetadata2.message()).thenReturn((V)"got message2"); - when(messageAndMetadata2.topic()).thenReturn("topic2"); - when(messageAndMetadata1.partition()).thenAnswer(getAnswer()); - - when(messageAndMetadata3.message()).thenReturn((V)"got message3"); - when(messageAndMetadata3.topic()).thenReturn("topic3"); - when(messageAndMetadata1.partition()).thenAnswer(getAnswer()); - - final Map>> messages = consumerConfiguration.receive(); - - int sum = 0; - - final Collection>> values = messages.values(); - - for (final Map> m : values) { - for (final List l : m.values()) { - sum += l.size(); - } - } - - Assert.assertEquals(9, sum); - } - - - - private Answer getAnswer() { - return new Answer() { - private int count = 0; - - @Override - public Object answer(final InvocationOnMock invocation) throws Throwable { - if (count++ == 1) { - return 1; - } else if (count++ == 2) { - return 2; - } - - return 3; - } - }; - } - - @Test - @SuppressWarnings("unchecked") - public void testReceiveMessageAndVerifyMessageLeftoverFromPreviousPollAreTakenFirst() { -<<<<<<< HEAD - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); -======= - final ConsumerMetadata consumerMetadata = Mockito.mock(ConsumerMetadata.class); ->>>>>>> INTEXT-40 - Add ZIP Transformer - final ConsumerConnectionProvider consumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); - final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); - - Map topicStreamMap = new HashMap(); - topicStreamMap.put("topic1", 1); - when(consumerMetadata.getTopicStreamMap()).thenReturn(topicStreamMap); - when(messageLeftOverTracker.getCurrentCount()).thenReturn(3); - - final MessageAndMetadata m1 = mock(MessageAndMetadata.class); - final MessageAndMetadata m2 = mock(MessageAndMetadata.class); - final MessageAndMetadata m3 = mock(MessageAndMetadata.class); - - when(m1.key()).thenReturn("key1"); - when(m1.message()).thenReturn("value1"); - when(m1.topic()).thenReturn("topic1"); - when(m1.partition()).thenReturn(1); - - when(m2.key()).thenReturn("key2"); - when(m2.message()).thenReturn("value2"); - when(m2.topic()).thenReturn("topic2"); - when(m2.partition()).thenReturn(1); - - when(m3.key()).thenReturn("key1"); - when(m3.message()).thenReturn("value3"); - when(m3.topic()).thenReturn("topic3"); - when(m3.partition()).thenReturn(1); - - final List> mList = new ArrayList>(); - mList.add(m1); - mList.add(m2); - mList.add(m3); - - when((List>) (Object) messageLeftOverTracker.getMessageLeftOverFromPreviousPoll()).thenReturn(mList); - - when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); - consumerConfiguration.setMaxMessages(5); - - final KafkaStream stream = mock(KafkaStream.class); - final List> streams = new ArrayList>(); - streams.add(stream); - final Map>> messageStreams = new HashMap>>(); - messageStreams.put("topic1", streams); - when(consumerConfiguration.createMessageStreamsForTopic()).thenReturn(messageStreams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); - when(stream.iterator()).thenReturn(iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); - when(iterator.next()).thenReturn(messageAndMetadata); - when(messageAndMetadata.message()).thenReturn((V) "got message"); - when(messageAndMetadata.topic()).thenReturn("topic1"); - when(messageAndMetadata.partition()).thenReturn(1); - - final Map>> messages = consumerConfiguration.receive(); - int sum = 0; - - final Collection>> values = messages.values(); - - for (final Map> m : values) { - for (final List l : m.values()) { - sum += l.size(); - } - - } - Assert.assertEquals(5, sum); - - Assert.assertTrue(messages.containsKey("topic1")); - Assert.assertTrue(messages.containsKey("topic2")); - Assert.assertTrue(messages.containsKey("topic3")); - - Assert.assertTrue(valueFound(messages.get("topic1").get(1), "value1")); - Assert.assertTrue(valueFound(messages.get("topic2").get(1), "value2")); - Assert.assertTrue(valueFound(messages.get("topic3").get(1), "value3")); - } - - @Test -<<<<<<< HEAD - @SuppressWarnings("unchecked") - public void testGetConsumerMapWithMessageStreamsWithNullDecoders() { - - final ConsumerMetadata mockedConsumerMetadata = mock(ConsumerMetadata.class); - - assertNull(mockedConsumerMetadata.getKeyDecoder()); - assertNull(mockedConsumerMetadata.getValueDecoder()); - - final Map topicsStreamMap = new HashMap(); - when(mockedConsumerMetadata.getTopicStreamMap()).thenReturn(topicsStreamMap); - - final ConsumerConnectionProvider mockedConsumerConnectionProvider = mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker mockedMessageLeftOverTracker = mock(MessageLeftOverTracker.class); - final ConsumerConnector mockedConsumerConnector = mock(ConsumerConnector.class); - - when(mockedConsumerConnectionProvider.getConsumerConnector()).thenReturn(mockedConsumerConnector); - - final Map>> messageStreams = new HashMap>>(); - when((Map>>) - (Object) mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(mockedConsumerMetadata, - mockedConsumerConnectionProvider, mockedMessageLeftOverTracker); - - consumerConfiguration.createMessageStreamsForTopic(); - - verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap(); - verify(mockedConsumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, null, null); - //verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap, null, null); - } - - @Test - @SuppressWarnings("unchecked") - public void testGetConsumerMapWithMessageStreamsWithDecoders() { - - @SuppressWarnings("unchecked") - final ConsumerMetadata mockedConsumerMetadata = mock(ConsumerMetadata.class); - - final Map topicsStreamMap = new HashMap(); - when(mockedConsumerMetadata.getTopicStreamMap()).thenReturn(topicsStreamMap); - - @SuppressWarnings("unchecked") - final Decoder mockedKeyDecoder = mock(Decoder.class); - - @SuppressWarnings("unchecked") - final Decoder mockedValueDecoder = mock(Decoder.class); - - when(mockedConsumerMetadata.getKeyDecoder()).thenReturn(mockedKeyDecoder); - when(mockedConsumerMetadata.getValueDecoder()).thenReturn(mockedValueDecoder); - - final ConsumerConnectionProvider mockedConsumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker mockedMessageLeftOverTracker = mock(MessageLeftOverTracker.class); - final ConsumerConnector mockedConsumerConnector = mock(ConsumerConnector.class); - - when(mockedConsumerConnectionProvider.getConsumerConnector()).thenReturn(mockedConsumerConnector); - - final Map>> messageStreams = new HashMap>>(); - when(mockedConsumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - - final ConsumerConfiguration consumerConfiguration = - new ConsumerConfiguration(mockedConsumerMetadata, mockedConsumerConnectionProvider, - mockedMessageLeftOverTracker); - - consumerConfiguration.createMessageStreamsForTopic(); - - verify(mockedConsumerMetadata, atLeast(1)).getTopicStreamMap(); - verify(mockedConsumerConnector, atMost(0)).createMessageStreams(topicsStreamMap); - verify(mockedConsumerConnector, atLeast(1)) - .createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder); - } - - - @Test - @SuppressWarnings("unchecked") - public void testReceiveMessageForTopicFilterFromSingleStream() { - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); - final ConsumerConnectionProvider consumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); - final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); - - when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); - - when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final ConsumerConfiguration consumerConfiguration = - new ConsumerConfiguration(consumerMetadata, consumerConnectionProvider, - messageLeftOverTracker); - consumerConfiguration.setMaxMessages(1); - - final KafkaStream stream = mock(KafkaStream.class); - final List> streams = new ArrayList>(); - streams.add(stream); - - when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); - final ConsumerIterator iterator = mock(ConsumerIterator.class); - when(stream.iterator()).thenReturn(iterator); - final MessageAndMetadata messageAndMetadata = mock(MessageAndMetadata.class); - when(iterator.next()).thenReturn(messageAndMetadata); - when(messageAndMetadata.message()).thenReturn("got message"); - when(messageAndMetadata.topic()).thenReturn("topic"); - when(messageAndMetadata.partition()).thenReturn(1); - - final Map>> messages = consumerConfiguration.receive(); - Assert.assertEquals(1, messages.size()); - Assert.assertEquals(1, messages.get("topic").size()); - Assert.assertEquals("got message", messages.get("topic").get(1).get(0)); - - verify(stream, times(1)).iterator(); - verify(iterator, times(1)).next(); - verify(messageAndMetadata, times(1)).message(); - verify(messageAndMetadata, times(1)).topic(); - } - - @Test - @SuppressWarnings("unchecked") - public void testReceiveMessageForTopicFilterFromMultipleStreams() { - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); - final ConsumerConnectionProvider consumerConnectionProvider = - mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = mock(MessageLeftOverTracker.class); - - when(consumerMetadata.getTopicFilterConfiguration()).thenReturn(new TopicFilterConfiguration(".*", 1, false)); - - final ConsumerConnector consumerConnector = mock(ConsumerConnector.class); - - when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final ConsumerConfiguration consumerConfiguration = - new ConsumerConfiguration(consumerMetadata, consumerConnectionProvider, - messageLeftOverTracker); - consumerConfiguration.setMaxMessages(3); - - final KafkaStream stream1 = mock(KafkaStream.class); - final KafkaStream stream2 = mock(KafkaStream.class); - final KafkaStream stream3 = mock(KafkaStream.class); - final List> streams = new ArrayList>(); - streams.add(stream1); - streams.add(stream2); - streams.add(stream3); - - when(consumerConfiguration.createMessageStreamsForTopicFilter()).thenReturn(streams); - final ConsumerIterator iterator1 = mock(ConsumerIterator.class); - final ConsumerIterator iterator2 = mock(ConsumerIterator.class); - final ConsumerIterator iterator3 = mock(ConsumerIterator.class); - - when(stream1.iterator()).thenReturn(iterator1); - when(stream2.iterator()).thenReturn(iterator2); - when(stream3.iterator()).thenReturn(iterator3); - final MessageAndMetadata messageAndMetadata1 = mock(MessageAndMetadata.class); - final MessageAndMetadata messageAndMetadata2 = mock(MessageAndMetadata.class); - final MessageAndMetadata messageAndMetadata3 = mock(MessageAndMetadata.class); - - when(iterator1.next()).thenReturn(messageAndMetadata1); - when(iterator2.next()).thenReturn(messageAndMetadata2); - when(iterator3.next()).thenReturn(messageAndMetadata3); - - when(messageAndMetadata1.message()).thenReturn("got message".getBytes()); - when(messageAndMetadata1.topic()).thenReturn("topic"); - when(messageAndMetadata1.partition()).thenReturn(1); - - when(messageAndMetadata2.message()).thenReturn("got message".getBytes()); - when(messageAndMetadata2.topic()).thenReturn("topic"); - when(messageAndMetadata2.partition()).thenReturn(2); - - when(messageAndMetadata3.message()).thenReturn("got message".getBytes()); - when(messageAndMetadata3.topic()).thenReturn("topic"); - when(messageAndMetadata3.partition()).thenReturn(3); - - final Map>> messages = consumerConfiguration.receive(); - Assert.assertEquals(1, messages.size()); - int sum = 0; - - final Map> values = messages.get("topic"); - - for (final List l : values.values()) { - sum += l.size(); - } - - Assert.assertEquals(3, sum); -======= - public void testGetConsumerMapWithMessageStreamsWithNullDecoders() { - - final ConsumerMetadata consumerMetadata = Mockito.mock(ConsumerMetadata.class); - - final Map topicsStreamMap = new HashMap(); - when(consumerMetadata.getTopicStreamMap()).thenReturn(topicsStreamMap); - - final ConsumerConnectionProvider consumerConnectionProvider = - Mockito.mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = Mockito.mock(MessageLeftOverTracker.class); - final ConsumerConnector consumerConnector = Mockito.mock(ConsumerConnector.class); - - Mockito.when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final Map>> messageStreams = new HashMap>>(); - Mockito.when(consumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - Mockito.when(consumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); - - consumerConfiguration.getConsumerMapWithMessageStreams(); - - verify(consumerMetadata, atLeast(1)).getTopicStreamMap(); - verify(consumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap); - verify(consumerConnector, atMost(0)).createMessageStreams(topicsStreamMap, null, null); - - } - - @Test - public void testGetConsumerMapWithMessageStreamsWithDecoders() { - - @SuppressWarnings("unchecked") - final ConsumerMetadata consumerMetadata = mock(ConsumerMetadata.class); - - final Map topicsStreamMap = new HashMap(); - when(consumerMetadata.getTopicStreamMap()).thenReturn(topicsStreamMap); - - @SuppressWarnings("unchecked") - final Decoder mockedKeyDecoder = (Decoder) mock(Decoder.class); - - @SuppressWarnings("unchecked") - final Decoder mockedValueDecoder = (Decoder) mock(Decoder.class); - - when(consumerMetadata.getKeyDecoder()).thenReturn(mockedKeyDecoder); - when(consumerMetadata.getValueDecoder()).thenReturn(mockedValueDecoder); - - final ConsumerConnectionProvider consumerConnectionProvider = - Mockito.mock(ConsumerConnectionProvider.class); - final MessageLeftOverTracker messageLeftOverTracker = Mockito.mock(MessageLeftOverTracker.class); - final ConsumerConnector consumerConnector = Mockito.mock(ConsumerConnector.class); - - Mockito.when(consumerConnectionProvider.getConsumerConnector()).thenReturn(consumerConnector); - - final Map>> messageStreams = new HashMap>>(); - Mockito.when(consumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - Mockito.when(consumerConnector.createMessageStreams(topicsStreamMap)).thenReturn(messageStreams); - - final ConsumerConfiguration consumerConfiguration = new ConsumerConfiguration(consumerMetadata, - consumerConnectionProvider, messageLeftOverTracker); - - consumerConfiguration.getConsumerMapWithMessageStreams(); - - verify(consumerMetadata, atLeast(1)).getTopicStreamMap(); - verify(consumerConnector, atMost(0)).createMessageStreams(topicsStreamMap); - verify(consumerConnector, atLeast(1)).createMessageStreams(topicsStreamMap, mockedKeyDecoder, mockedValueDecoder); - ->>>>>>> INTEXT-40 - Add ZIP Transformer - } - - private boolean valueFound(final List l, final String value){ - for (final Object o : l){ - if (value.equals(o)){ - return true; - } - } - - return false; - } -}