Fix sonar violation

Fix sonar violation'

Change-Id: Ia5718d2bcbf9f5efea40d8250b7ad57f6d2eb2f3
Issue-ID: DCAEGEN2-1016
Signed-off-by: Zlatko Murgoski <zlatko.murgoski@nokia.com>

6 files changed, 524 insertions, 0 deletions

diff --git a/src/test/java/org/onap/dcae/common/ApiExceptionTest.java b/src/test/java/org/onap/dcae/common/ApiExceptionTest.java
new file mode 100644
index 00000000..2436694c
--- /dev/null
+++ b/src/test/java/org/onap/dcae/common/ApiExceptionTest.java
@@ -0,0 +1,60 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcae.common;
+
+import static org.junit.Assert.assertEquals;
+
+import org.json.JSONObject;
+import org.junit.Test;
+import org.onap.dcae.restapi.ApiException;
+import org.onap.dcae.restapi.ApiException.ExceptionType;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public class ApiExceptionTest {
+
+    @Test
+    public void shouldStringifyServiceExceptionTypeAccordingToSpecification() {
+        assertEquals(ExceptionType.SERVICE_EXCEPTION.toString(), "ServiceException");
+    }
+
+    @Test
+    public void shouldStringifyPolicyExceptionTypeAccordingToSpecification() {
+        assertEquals(ExceptionType.POLICY_EXCEPTION.toString(), "PolicyException");
+    }
+
+    @Test
+    public void shouldConvertExceptionToBackwardCompatibleFormat() {
+        JSONObject responseBody = ApiException.UNAUTHORIZED_USER.toJSON();
+        assertEquals(responseBody.toString(), new JSONObject(""
+            + "{                                             "
+            + "  'requestError': {                           "
+            + "     'PolicyException': {                     "
+            + "        'messageId': 'POL2000',                "
+            + "        'text': 'Unauthorized user'           "
+            + "     }                                        "
+            + "  }                                           "
+            + "}                                             "
+            .replace("'", "\"")
+        ).toString());
+    }
+}
diff --git a/src/test/java/org/onap/dcae/common/ConfigProcessorAdapterTest.java b/src/test/java/org/onap/dcae/common/ConfigProcessorAdapterTest.java
new file mode 100644
index 00000000..214badaf
--- /dev/null
+++ b/src/test/java/org/onap/dcae/common/ConfigProcessorAdapterTest.java
@@ -0,0 +1,66 @@
+//
+//        ================================================================================
+//        Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
+//        Copyright (c) 2018 Nokia. All rights reserved.
+//        ================================================================================
+//        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.
+//        ============LICENSE_END=========================================================
+//
+//
+package org.onap.dcae.common;
+
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.json.JSONObject;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ConfigProcessorAdapterTest {
+
+    @Mock
+    private ConfigProcessors configProcessors;
+
+    @InjectMocks
+    private ConfigProcessorAdapter configProcessorAdapter;
+
+
+    @Test
+    public void shouldCallIsFilterMetOnAdapter() {
+        //given
+        JSONObject parameter = new JSONObject();
+        when(configProcessors.isFilterMet(parameter)).thenReturn(true);
+        //when
+        boolean actualReturn = configProcessorAdapter.isFilterMet(parameter);
+        //then
+        assertTrue(actualReturn);
+        verify(configProcessors, times(1)).isFilterMet(parameter);
+    }
+
+    @Test
+    public void shouldCallGivenMethodFromConfigProcessor() throws Exception {
+        JSONObject parameter = new JSONObject();
+        String exampleFunction = "concatenateValue";
+        //when
+        configProcessorAdapter.runConfigProcessorFunctionByName(exampleFunction, parameter);
+        //then
+        verify(configProcessors, times(1)).concatenateValue(parameter);
+    }
+
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/common/EventSenderTest.java b/src/test/java/org/onap/dcae/common/EventSenderTest.java
new file mode 100644
index 00000000..aba3c2a9
--- /dev/null
+++ b/src/test/java/org/onap/dcae/common/EventSenderTest.java
@@ -0,0 +1,69 @@
+/*
+ * ============LICENSE_START=======================================================
+ * PROJECT
+ * ================================================================================
+ * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.s
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.common;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.vavr.collection.HashMap;
+import io.vavr.collection.Map;
+import org.json.JSONObject;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.onap.dcae.ApplicationSettings;
+import org.onap.dcae.common.publishing.EventPublisher;
+
+@RunWith(MockitoJUnitRunner.Silent.class)
+public class EventSenderTest {
+
+
+  private String event = "{\"VESversion\":\"v7\",\"VESuniqueId\":\"fd69d432-5cd5-4c15-9d34-407c81c61c6a-0\",\"event\":{\"commonEventHeader\":{\"startEpochMicrosec\":1544016106000000,\"eventId\":\"fault33\",\"timeZoneOffset\":\"UTC+00.00\",\"priority\":\"Normal\",\"version\":\"4.0.1\",\"nfVendorName\":\"Ericsson\",\"reportingEntityName\":\"1\",\"sequence\":1,\"domain\":\"fault\",\"lastEpochMicrosec\":1544016106000000,\"eventName\":\"Fault_KeyFileFault\",\"vesEventListenerVersion\":\"7.0.1\",\"sourceName\":\"1\"},\"faultFields\":{\"eventSeverity\":\"CRITICAL\",\"alarmCondition\":\"KeyFileFault\",\"faultFieldsVersion\":\"4.0\",\"eventCategory\":\"PROCESSINGERRORALARM\",\"specificProblem\":\"License Key File Fault_1\",\"alarmAdditionalInformation\":{\"probableCause\":\"ConfigurationOrCustomizationError\",\"additionalText\":\"test_1\",\"source\":\"ManagedElement=1,SystemFunctions=1,Lm=1\"},\"eventSourceType\":\"Lm\",\"vfStatus\":\"Active\"}}}\n";
+
+  @Mock
+  private EventPublisher eventPublisher;
+  @Mock
+  private ApplicationSettings settings;
+
+  private EventSender eventSender;
+
+
+  @Test
+  public void shouldntSendEventWhenStreamIdsIsEmpty() {
+    when(settings.dMaaPStreamsMapping()).thenReturn(HashMap.empty());
+    eventSender = new EventSender(eventPublisher, settings );
+    eventSender.send(new JSONObject(event));
+    verify(eventPublisher,never()).sendEvent(any(),any());
+  }
+
+  @Test
+  public void shouldSendEvent() {
+    Map<String, String[]> streams = HashMap.of("fault", new String[]{"ves-fault", "fault-ves"});
+    when(settings.dMaaPStreamsMapping()).thenReturn(streams);
+    eventSender = new EventSender(eventPublisher, settings );
+    eventSender.send(new JSONObject(event));
+    verify(eventPublisher, times(2)).sendEvent(any(),any());
+  }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/common/publishing/DMaaPConfigurationParserTest.java b/src/test/java/org/onap/dcae/common/publishing/DMaaPConfigurationParserTest.java
new file mode 100644
index 00000000..923aae02
--- /dev/null
+++ b/src/test/java/org/onap/dcae/common/publishing/DMaaPConfigurationParserTest.java
@@ -0,0 +1,114 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.common.publishing;
+
+import static io.vavr.API.List;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.onap.dcae.common.publishing.DMaaPConfigurationParser.parseToDomainMapping;
+
+import io.vavr.collection.Map;
+import io.vavr.control.Try;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import org.junit.Test;
+
+/**
+ * @author Pawel Szalapski (pawel.szalapski@nokia.com)
+ */
+public class DMaaPConfigurationParserTest {
+
+    @Test
+    public void testParseCredentialsForGen2() {
+        Path path = Paths.get("src/test/resources/testParseDMaaPCredentialsGen2.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+        PublisherConfig authCredentialsNulls = publisherConfigs.get().get("auth-credentials-null").getOrNull();
+        assertThat(authCredentialsNulls.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsNulls.password().isEmpty()).isTrue();
+        assertThat(authCredentialsNulls.isSecured()).isFalse();
+
+        PublisherConfig authCredentialsPresent = publisherConfigs.get().get("auth-credentials-present").getOrNull();
+        assertThat(authCredentialsPresent.userName().getOrNull()).isEqualTo("sampleUser");
+        assertThat(authCredentialsPresent.password().getOrNull()).isEqualTo("samplePassword");
+        assertThat(authCredentialsPresent.isSecured()).isTrue();
+
+        PublisherConfig authCredentialsKeysMissing = publisherConfigs.get().get("auth-credentials-missing").getOrNull();
+        assertThat(authCredentialsKeysMissing.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsKeysMissing.password().isEmpty()).isTrue();
+        assertThat(authCredentialsKeysMissing.isSecured()).isFalse();
+    }
+
+
+    @Test
+    public void testParseCredentialsForLegacy() {
+        Path path = Paths.get("src/test/resources/testParseDMaaPCredentialsLegacy.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+        PublisherConfig authCredentialsNull = publisherConfigs.get().get("auth-credentials-null").getOrNull();
+        assertThat(authCredentialsNull.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsNull.password().isEmpty()).isTrue();
+        assertThat(authCredentialsNull.isSecured()).isFalse();
+
+        PublisherConfig authCredentialsPresent = publisherConfigs.get().get("auth-credentials-present").getOrNull();
+        assertThat(authCredentialsPresent.userName().getOrNull()).isEqualTo("sampleUser");
+        assertThat(authCredentialsPresent.password().getOrNull()).isEqualTo("samplePassword");
+        assertThat(authCredentialsPresent.isSecured()).isTrue();
+
+        PublisherConfig authCredentialsMissing = publisherConfigs.get().get("auth-credentials-missing").getOrNull();
+        assertThat(authCredentialsMissing.userName().isEmpty()).isTrue();
+        assertThat(authCredentialsMissing.password().isEmpty()).isTrue();
+        assertThat(authCredentialsMissing.isSecured()).isFalse();
+    }
+
+
+    @Test
+    public void testParseGen2() {
+        Path path = Paths.get("src/test/resources/testParseDMaaPGen2.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = parseToDomainMapping(path);
+
+        PublisherConfig withEventsSegment = publisherConfigs.get().get("event-segments-with-port").getOrNull();
+        assertThat(withEventsSegment.destinations()).isEqualTo(List("UEBHOST:3904"));
+        assertThat(withEventsSegment.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+
+        PublisherConfig withOtherSegment = publisherConfigs.get().get("other-segments-without-ports").getOrNull();
+        assertThat(withOtherSegment.destinations()).isEqualTo(List("UEBHOST"));
+        assertThat(withOtherSegment.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+    }
+
+    @Test
+    public void testParseLegacy() {
+        Path exemplaryConfig = Paths.get("src/test/resources/testParseDMaaPLegacy.json");
+        Try<Map<String, PublisherConfig>> publisherConfigs = DMaaPConfigurationParser
+            .parseToDomainMapping(exemplaryConfig);
+
+        PublisherConfig urlFirstThenHosts = publisherConfigs.get().get("url-precedes-hosts").getOrNull();
+        assertThat(urlFirstThenHosts.destinations()).isEqualTo(List("127.0.0.1:3904"));
+        assertThat(urlFirstThenHosts.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+
+        PublisherConfig urlKeyMissing = publisherConfigs.get().get("url-key-missing").getOrNull();
+        assertThat(urlKeyMissing.destinations()).isEqualTo(List("h1.att.com", "h2.att.com"));
+        assertThat(urlKeyMissing.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+
+        PublisherConfig urlIsMissing = publisherConfigs.get().get("url-is-null").getOrNull();
+        assertThat(urlIsMissing.destinations()).isEqualTo(List("h1.att.com", "h2.att.com"));
+        assertThat(urlIsMissing.topic()).isEqualTo("DCAE-SE-COLLECTOR-EVENTS-DEV");
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java b/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java
new file mode 100644
index 00000000..809ac99c
--- /dev/null
+++ b/src/test/java/org/onap/dcae/common/publishing/DMaaPEventPublisherTest.java
@@ -0,0 +1,89 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.common.publishing;
+
+import static io.vavr.API.Option;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import java.io.IOException;
+import org.json.JSONObject;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+
+public class DMaaPEventPublisherTest {
+
+    private static final String STREAM_ID = "sampleStreamId";
+
+    private DMaaPEventPublisher eventPublisher;
+    private CambriaBatchingPublisher cambriaPublisher;
+    private DMaaPPublishersCache DMaaPPublishersCache;
+
+    @Before
+    public void setUp() {
+        cambriaPublisher = mock(CambriaBatchingPublisher.class);
+        DMaaPPublishersCache = mock(DMaaPPublishersCache.class);
+        when(DMaaPPublishersCache.getPublisher(anyString())).thenReturn(Option(cambriaPublisher));
+        eventPublisher = new DMaaPEventPublisher(DMaaPPublishersCache, mock(Logger.class));
+    }
+
+    @Test
+    public void shouldSendEventToTopic() throws Exception {
+        // given
+        JSONObject event = new JSONObject("{}");
+
+        // when
+        eventPublisher.sendEvent(event, STREAM_ID);
+
+        // then
+        verify(cambriaPublisher).send("MyPartitionKey", event.toString());
+    }
+
+    @Test
+    public void shouldRemoveInternalVESUIDBeforeSending() throws Exception {
+        // given
+        JSONObject event = new JSONObject(
+            "{\"VESuniqueId\": \"362e0146-ec5f-45f3-8d8f-bfe877c3f58e\", \"another\": 8}");
+
+        // when
+        eventPublisher.sendEvent(event, STREAM_ID);
+
+        // then
+        verify(cambriaPublisher).send("MyPartitionKey", new JSONObject("{\"another\": 8}").toString());
+    }
+
+    @Test
+    public void shouldCloseConnectionWhenExceptionOccurred() throws Exception {
+        // given
+        JSONObject event = new JSONObject("{}");
+        given(cambriaPublisher.send(anyString(), anyString())).willThrow(new IOException("epic fail"));
+
+        // when
+        eventPublisher.sendEvent(event, STREAM_ID);
+
+        // then
+        verify(DMaaPPublishersCache).closePublisherFor(STREAM_ID);
+    }
+}
\ No newline at end of file
diff --git a/src/test/java/org/onap/dcae/common/publishing/DMaaPPublishersCacheTest.java b/src/test/java/org/onap/dcae/common/publishing/DMaaPPublishersCacheTest.java
new file mode 100644
index 00000000..f4dbe190
--- /dev/null
+++ b/src/test/java/org/onap/dcae/common/publishing/DMaaPPublishersCacheTest.java
@@ -0,0 +1,126 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * org.onap.dcaegen2.collectors.ves
+ * ================================================================================
+ * Copyright (C) 2018 Nokia. All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.dcae.common.publishing;
+
+import static io.vavr.API.List;
+import static io.vavr.API.Map;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.mockito.Mockito.when;
+
+import com.att.nsa.cambria.client.CambriaBatchingPublisher;
+import io.vavr.collection.Map;
+import io.vavr.control.Option;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.dcae.common.publishing.DMaaPPublishersCache.CambriaPublishersCacheLoader;
+import org.onap.dcae.common.publishing.DMaaPPublishersCache.OnPublisherRemovalListener;
+
+
+public class DMaaPPublishersCacheTest {
+
+    private String streamId1;
+    private Map<String, PublisherConfig> dMaaPConfigs;
+
+    @Before
+    public void setUp() {
+        streamId1 = "sampleStream1";
+        dMaaPConfigs = Map("sampleStream1", new PublisherConfig(List("destination1"), "topic1"));
+    }
+
+    @Test
+    public void shouldReturnTheSameCachedInstanceOnConsecutiveRetrievals() {
+        // given
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(dMaaPConfigs);
+
+        // when
+        Option<CambriaBatchingPublisher> firstPublisher = dMaaPPublishersCache.getPublisher(streamId1);
+        Option<CambriaBatchingPublisher> secondPublisher = dMaaPPublishersCache.getPublisher(streamId1);
+
+        // then
+        assertSame("should return same instance", firstPublisher.get(), secondPublisher.get());
+    }
+
+    @Test
+    public void shouldCloseCambriaPublisherOnCacheInvalidate() throws IOException, InterruptedException {
+        // given
+        CambriaBatchingPublisher cambriaPublisherMock1 = mock(CambriaBatchingPublisher.class);
+        CambriaPublishersCacheLoader cacheLoaderMock = mock(CambriaPublishersCacheLoader.class);
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(cacheLoaderMock,
+                                                                             new OnPublisherRemovalListener(),
+                                                                             dMaaPConfigs);
+        when(cacheLoaderMock.load(streamId1)).thenReturn(cambriaPublisherMock1);
+
+        // when
+        dMaaPPublishersCache.getPublisher(streamId1);
+        dMaaPPublishersCache.closePublisherFor(streamId1);
+
+        // then
+        verify(cambriaPublisherMock1).close(20, TimeUnit.SECONDS);
+
+    }
+
+    @Test
+    public void shouldReturnNoneIfThereIsNoDMaaPConfigurationForGivenStreamID() {
+        // given
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(dMaaPConfigs);
+
+        // then
+        assertTrue("should not exist", dMaaPPublishersCache.getPublisher("non-existing").isEmpty());
+    }
+
+
+    @Test
+    public void shouldCloseOnlyChangedPublishers() throws IOException, InterruptedException {
+        // given
+        CambriaBatchingPublisher cambriaPublisherMock1 = mock(CambriaBatchingPublisher.class);
+        CambriaBatchingPublisher cambriaPublisherMock2 = mock(CambriaBatchingPublisher.class);
+        CambriaPublishersCacheLoader cacheLoaderMock = mock(CambriaPublishersCacheLoader.class);
+        String firstDomain = "domain1";
+        String secondDomain = "domain2";
+        Map<String, PublisherConfig> oldConfig = Map(firstDomain,
+                                                     new PublisherConfig(List("destination1"), "topic1"),
+                                                     secondDomain,
+                                                     new PublisherConfig(List("destination2"), "topic2",
+                                                                         "user", "pass"));
+        Map<String, PublisherConfig> newConfig = Map(firstDomain, new PublisherConfig(List("destination1"), "topic1"),
+                                                     secondDomain, new PublisherConfig(List("destination2"), "topic2"));
+        DMaaPPublishersCache dMaaPPublishersCache = new DMaaPPublishersCache(cacheLoaderMock,
+                                                                             new OnPublisherRemovalListener(),
+                                                                             oldConfig);
+        when(cacheLoaderMock.load(firstDomain)).thenReturn(cambriaPublisherMock1);
+        when(cacheLoaderMock.load(secondDomain)).thenReturn(cambriaPublisherMock2);
+
+        dMaaPPublishersCache.getPublisher(firstDomain);
+        dMaaPPublishersCache.getPublisher(secondDomain);
+
+        // when
+        dMaaPPublishersCache.reconfigure(newConfig);
+
+        // then
+        verify(cambriaPublisherMock2).close(20, TimeUnit.SECONDS);
+        verifyZeroInteractions(cambriaPublisherMock1);
+    }
+}
\ No newline at end of file