1 files changed, 193 insertions, 153 deletions
diff --git a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
index 0255c100..2c33a257 100644
--- a/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
+++ b/policy-endpoints/src/test/java/org/onap/policy/common/endpoints/event/comm/bus/internal/BusConsumerTest.java
@@ -2,7 +2,8 @@
  * ============LICENSE_START=======================================================
  * policy-endpoints
  * ================================================================================
- * Copyright (C) 2018-2019 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018-2021 AT&T Intellectual Property. All rights reserved.
+ * Modifications Copyright (C) 2023-2024 Nordix Foundation.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,218 +21,257 @@
 
 package org.onap.policy.common.endpoints.event.comm.bus.internal;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
-import static org.mockito.Mockito.mock;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
-import com.att.nsa.cambria.client.CambriaConsumer;
 import java.io.IOException;
-import java.util.Arrays;
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
-import org.apache.commons.collections4.IteratorUtils;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import org.onap.dmaap.mr.client.impl.MRConsumerImpl;
-import org.onap.dmaap.mr.client.response.MRConsumerResponse;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
 import org.onap.policy.common.endpoints.event.comm.bus.TopicTestBase;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.CambriaConsumerWrapper;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapAafConsumerWrapper;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapConsumerWrapper;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.DmaapDmeConsumerWrapper;
-import org.powermock.reflect.Whitebox;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.FetchingBusConsumer;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.KafkaConsumerWrapper;
+import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
 
 public class BusConsumerTest extends TopicTestBase {
 
+    private static final int SHORT_TIMEOUT_MILLIS = 10;
+    private static final int LONG_TIMEOUT_MILLIS = 3000;
+
+    @Mock
+    KafkaConsumer<String, String> mockedKafkaConsumer;
+
+    AutoCloseable closeable;
+
     @Before
     @Override
     public void setUp() {
         super.setUp();
+        closeable = MockitoAnnotations.openMocks(this);
     }
 
-    @Test
-    public void testCambriaConsumerWrapper() {
-        // verify that different wrappers can be built
-        new CambriaConsumerWrapper(makeBuilder().build());
-        new CambriaConsumerWrapper(makeBuilder().useHttps(false).build());
-        new CambriaConsumerWrapper(makeBuilder().useHttps(true).build());
-        new CambriaConsumerWrapper(makeBuilder().useHttps(true).allowSelfSignedCerts(false).build());
-        new CambriaConsumerWrapper(makeBuilder().useHttps(true).allowSelfSignedCerts(true).build());
-        new CambriaConsumerWrapper(makeBuilder().apiKey(null).build());
-        new CambriaConsumerWrapper(makeBuilder().apiSecret(null).build());
-        new CambriaConsumerWrapper(makeBuilder().apiKey(null).apiSecret(null).build());
-        new CambriaConsumerWrapper(makeBuilder().userName(null).build());
-        new CambriaConsumerWrapper(makeBuilder().password(null).build());
-        new CambriaConsumerWrapper(makeBuilder().userName(null).password(null).build());
+    @After
+    public void tearDown() throws Exception {
+        closeable.close();
     }
 
-    @Test
-    public void testCambriaConsumerWrapperFetch() throws Exception {
-        CambriaConsumer inner = mock(CambriaConsumer.class);
-        List<String> lst = Arrays.asList(MY_MESSAGE, MY_MESSAGE2);
-        when(inner.fetch()).thenReturn(lst);
-
-        CambriaConsumerWrapper cons = new CambriaConsumerWrapper(builder.build());
-        Whitebox.setInternalState(cons, "consumer", inner);
-
-        assertEquals(lst, IteratorUtils.toList(cons.fetch().iterator()));
 
-        // arrange to throw exception next time fetch is called
-        IOException ex = new IOException(EXPECTED);
-        when(inner.fetch()).thenThrow(ex);
-
-        cons.fetchTimeout = 10;
-
-        try {
-            cons.fetch();
-            fail("missing exception");
-
-        } catch (IOException e) {
-            assertEquals(ex, e);
-        }
+    @Test
+    public void testFetchingBusConsumer() {
+        // should not be negative
+        var cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(-1).build());
+        assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH);
+
+        // should not be zero
+        cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(0).build());
+        assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH);
+
+        // should not be too large
+        cons = new FetchingBusConsumerImpl(
+                        makeBuilder().fetchTimeout(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH + 100).build());
+        assertThat(cons.getSleepTime()).isEqualTo(PolicyEndPointProperties.DEFAULT_TIMEOUT_MS_FETCH);
+
+        // should not be what was specified
+        cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(100).build());
+        assertThat(cons.getSleepTime()).isEqualTo(100);
     }
 
     @Test
-    public void testCambriaConsumerWrapperClose() {
-        CambriaConsumerWrapper cons = new CambriaConsumerWrapper(builder.build());
-
-        // set filter several times to cause different branches of close() to be executed
-        for (int count = 0; count < 3; ++count) {
-            cons.close();
-            cons.setFilter("close=" + count);
-        }
+    public void testFetchingBusConsumerSleepAfterFetchFailure() throws InterruptedException {
+
+        var cons = new FetchingBusConsumerImpl(makeBuilder().fetchTimeout(SHORT_TIMEOUT_MILLIS).build()) {
+
+            private CountDownLatch started = new CountDownLatch(1);
+
+            @Override
+            protected void sleepAfterFetchFailure() {
+                started.countDown();
+                super.sleepAfterFetchFailure();
+            }
+        };
+
+        // full sleep
+        long tstart = System.currentTimeMillis();
+        cons.sleepAfterFetchFailure();
+        assertThat(System.currentTimeMillis() - tstart).isGreaterThanOrEqualTo(SHORT_TIMEOUT_MILLIS);
+
+        // close while sleeping - sleep should halt prematurely
+        cons.fetchTimeout = LONG_TIMEOUT_MILLIS;
+        cons.started = new CountDownLatch(1);
+        Thread thread = new Thread(cons::sleepAfterFetchFailure);
+        tstart = System.currentTimeMillis();
+        thread.start();
+        cons.started.await();
+        cons.close();
+        thread.join();
+        assertThat(System.currentTimeMillis() - tstart).isLessThan(LONG_TIMEOUT_MILLIS);
+
+        // interrupt while sleeping - sleep should halt prematurely
+        cons.fetchTimeout = LONG_TIMEOUT_MILLIS;
+        cons.started = new CountDownLatch(1);
+        thread = new Thread(cons::sleepAfterFetchFailure);
+        tstart = System.currentTimeMillis();
+        thread.start();
+        cons.started.await();
+        thread.interrupt();
+        thread.join();
+        assertThat(System.currentTimeMillis() - tstart).isLessThan(LONG_TIMEOUT_MILLIS);
     }
 
     @Test
-    public void testCambriaConsumerWrapperSetFilter() {
-        // set filter several times to cause different branches to be executed
-        CambriaConsumerWrapper cons = new CambriaConsumerWrapper(builder.build());
-        for (int count = 0; count < 3; ++count) {
-            cons.setFilter("set-filter=" + count);
-        }
+    public void testKafkaConsumerWrapper() {
+        // verify that different wrappers can be built
+        assertThatCode(() -> new KafkaConsumerWrapper(makeKafkaBuilder().build())).doesNotThrowAnyException();
     }
 
-    @Test
-    public void testCambriaConsumerWrapperToString() {
-        assertNotNull(new CambriaConsumerWrapper(makeBuilder().build()).toString());
+    @Test(expected = IllegalArgumentException.class)
+    public void testKafkaConsumerWrapper_InvalidTopic() {
+        new KafkaConsumerWrapper(makeBuilder().topic(null).build());
     }
 
     @Test
-    public void testDmaapConsumerWrapper() throws Exception {
-        // verify that different wrappers can be built
-        new DmaapAafConsumerWrapper(makeBuilder().build());
+    public void testKafkaConsumerWrapperFetch() {
+
+        //Setup Properties for consumer
+        Properties kafkaProps = new Properties();
+        kafkaProps.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+        kafkaProps.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test");
+        kafkaProps.setProperty("enable.auto.commit", "true");
+        kafkaProps.setProperty("auto.commit.interval.ms", "1000");
+        kafkaProps.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+            "org.apache.kafka.common.serialization.StringDeserializer");
+        kafkaProps.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+            "org.apache.kafka.common.serialization.StringDeserializer");
+        kafkaProps.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        kafkaProps.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
+
+        KafkaConsumerWrapper kafka = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kafkaProps);
+        kafka.consumer = consumer;
+
+        assertThrows(java.lang.IllegalStateException.class, () -> kafka.fetch().iterator().hasNext());
+        consumer.close();
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapConsumerWrapper_InvalidTopic() throws Exception {
-        new DmaapAafConsumerWrapper(makeBuilder().topic(null).build());
+    @Test
+    public void testFetchNoMessages() throws IOException {
+        KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+        kafkaConsumerWrapper.consumer = mockedKafkaConsumer;
+
+        when(mockedKafkaConsumer.poll(any())).thenReturn(new ConsumerRecords<>(Collections.emptyMap()));
+
+        Iterable<String> result = kafkaConsumerWrapper.fetch();
+
+        verify(mockedKafkaConsumer, times(1)).poll(any());
+
+        assertThat(result != null);
+
+        assertThat(!result.iterator().hasNext());
+
+        mockedKafkaConsumer.close();
     }
 
     @Test
-    public void testDmaapConsumerWrapperFetch() throws Exception {
-        DmaapAafConsumerWrapper dmaap = new DmaapAafConsumerWrapper(makeBuilder().build());
-        MRConsumerImpl cons = mock(MRConsumerImpl.class);
+    public void testFetchWithMessages() {
+        // Setup
+        KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+        kafkaConsumerWrapper.consumer = mockedKafkaConsumer;
 
-        dmaap.fetchTimeout = 5;
-        dmaap.consumer = cons;
+        ConsumerRecord<String, String> record = new ConsumerRecord<>("my-effective-topic", 0, 0, "key", "value");
+        Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<>();
+        recordsMap.put(new TopicPartition("my-effective-topic", 0), Collections.singletonList(record));
+        ConsumerRecords<String, String> consumerRecords = new ConsumerRecords<>(recordsMap);
 
-        // null return
-        when(cons.fetchWithReturnConsumerResponse()).thenReturn(null);
-        assertFalse(dmaap.fetch().iterator().hasNext());
+        when(mockedKafkaConsumer.poll(any())).thenReturn(consumerRecords);
 
-        // with messages, 200
-        List<String> lst = Arrays.asList(MY_MESSAGE, MY_MESSAGE2);
-        MRConsumerResponse resp = new MRConsumerResponse();
-        resp.setResponseCode("200");
-        resp.setActualMessages(lst);
-        when(cons.fetchWithReturnConsumerResponse()).thenReturn(resp);
+        Iterable<String> result = kafkaConsumerWrapper.fetch();
 
-        assertEquals(lst, IteratorUtils.toList(dmaap.fetch().iterator()));
+        verify(mockedKafkaConsumer, times(1)).poll(any());
 
-        // null messages
-        resp.setActualMessages(null);
-        when(cons.fetchWithReturnConsumerResponse()).thenReturn(resp);
+        verify(mockedKafkaConsumer, times(1)).commitSync(any(Map.class));
 
-        assertFalse(dmaap.fetch().iterator().hasNext());
+        assertThat(result != null);
 
-        // with messages, NOT 200
-        resp.setResponseCode("400");
-        resp.setActualMessages(lst);
-        when(cons.fetchWithReturnConsumerResponse()).thenReturn(resp);
+        assertThat(result.iterator().hasNext());
 
-        assertEquals(lst, IteratorUtils.toList(dmaap.fetch().iterator()));
-    }
+        assertThat(result.iterator().next().equals("value"));
 
-    @Test
-    public void testDmaapConsumerWrapperClose() throws Exception {
-        new DmaapAafConsumerWrapper(makeBuilder().build()).close();
+        mockedKafkaConsumer.close();
     }
 
     @Test
-    public void testDmaapConsumerWrapperToString() throws Exception {
-        assertNotNull(new DmaapConsumerWrapper(makeBuilder().build()) {}.toString());
-    }
+    public void testFetchWithMessagesAndTraceparent() {
+        // Setup
+        KafkaConsumerWrapper kafkaConsumerWrapper = new KafkaConsumerWrapper(makeKafkaBuilder().build());
+        kafkaConsumerWrapper.consumer = mockedKafkaConsumer;
 
-    @Test
-    public void testDmaapAafConsumerWrapper() throws Exception {
-        // verify that different wrappers can be built
-        new DmaapAafConsumerWrapper(makeBuilder().useHttps(true).build());
-        new DmaapAafConsumerWrapper(makeBuilder().useHttps(false).build());
-    }
+        ConsumerRecord<String, String> record = new ConsumerRecord<>("my-effective-topic", 0, 0, "key", "value");
+        record.headers().add(
+                "traceparent",
+                "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01".getBytes(StandardCharsets.UTF_8)
+        );
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapAafConsumerWrapper_InvalidServers() throws Exception {
-        /*
-         * Unfortunately, the MR code intercepts this and throws an exception before the
-         * wrapper gets a chance to check it, thus this test does not improve the coverage
-         * for the constructor.
-         */
-        new DmaapAafConsumerWrapper(makeBuilder().servers(Collections.emptyList()).build());
-    }
+        Map<TopicPartition, List<ConsumerRecord<String, String>>> recordsMap = new HashMap<>();
+        recordsMap.put(new TopicPartition("my-effective-topic", 0), Collections.singletonList(record));
+        ConsumerRecords<String, String> consumerRecords = new ConsumerRecords<>(recordsMap);
 
-    @Test
-    public void testDmaapAafConsumerWrapperToString() throws Exception {
-        assertNotNull(new DmaapAafConsumerWrapper(makeBuilder().build()).toString());
-    }
+        when(mockedKafkaConsumer.poll(any())).thenReturn(consumerRecords);
 
-    @Test
-    public void testDmaapDmeConsumerWrapper() throws Exception {
-        // verify that different wrappers can be built
-        new DmaapDmeConsumerWrapper(makeBuilder().build());
-        new DmaapDmeConsumerWrapper(makeBuilder().useHttps(true).build());
-        new DmaapDmeConsumerWrapper(makeBuilder().useHttps(false).build());
-        new DmaapDmeConsumerWrapper(makeBuilder().additionalProps(null).build());
-
-        addProps.put(ROUTE_PROP, MY_ROUTE);
-        new DmaapDmeConsumerWrapper(makeBuilder().build());
-        new DmaapDmeConsumerWrapper(makeBuilder().partner(null).build());
-    }
+        Iterable<String> result = kafkaConsumerWrapper.fetch();
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapDmeConsumerWrapper_InvalidEnvironment() throws Exception {
-        new DmaapDmeConsumerWrapper(makeBuilder().environment(null).build());
-    }
+        verify(mockedKafkaConsumer, times(1)).poll(any());
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapDmeConsumerWrapper_InvalidAft() throws Exception {
-        new DmaapDmeConsumerWrapper(makeBuilder().aftEnvironment(null).build());
+        verify(mockedKafkaConsumer, times(1)).commitSync(any(Map.class));
+
+        assertThat(result != null);
+
+        assertThat(result.iterator().hasNext());
+
+        assertThat(result.iterator().next().equals("value"));
+
+        mockedKafkaConsumer.close();
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapDmeConsumerWrapper_InvalidLat() throws Exception {
-        new DmaapDmeConsumerWrapper(makeBuilder().latitude(null).build());
+
+    @Test
+    public void testKafkaConsumerWrapperClose() {
+        assertThatCode(() -> new KafkaConsumerWrapper(makeKafkaBuilder().build()).close()).doesNotThrowAnyException();
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapDmeConsumerWrapper_InvalidLong() throws Exception {
-        new DmaapDmeConsumerWrapper(makeBuilder().longitude(null).build());
+    @Test
+    public void testKafkaConsumerWrapperToString() {
+        assertNotNull(new KafkaConsumerWrapper(makeKafkaBuilder().build()) {}.toString());
     }
 
-    @Test(expected = IllegalArgumentException.class)
-    public void testDmaapDmeConsumerWrapper_InvalidPartner() throws Exception {
-        new DmaapDmeConsumerWrapper(makeBuilder().partner(null).build());
+    private static class FetchingBusConsumerImpl extends FetchingBusConsumer {
+
+        protected FetchingBusConsumerImpl(BusTopicParams busTopicParams) {
+            super(busTopicParams);
+        }
+
+        @Override
+        public Iterable<String> fetch() {
+            return null;
+        }
     }
 }