* 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=========================================================
- *
+ *
* ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
+ *
*******************************************************************************/
package org.onap.dmaap.dmf.mr.backends.kafka;
// check for expirations pretty regularly
private static final long kDefault_SweepEverySeconds = 15;
- private enum Status {
- NOT_STARTED, CONNECTED, DISCONNECTED, SUSPENDED
- }
-
-
+ // private final rrNvReadable fSettings;
+ private MetricsSet fMetrics;
+ private final String fBaseZkPath;
+ private final ScheduledExecutorService fSweepScheduler;
+ private String fApiId;
+
+ private final ConnectionStateListener listener;
+
+ private ConcurrentHashMap<String, Kafka011Consumer> fConsumers;
+ private PathChildrenCache curatorConsumerCache;
+
+ private volatile Status status;
@Autowired
private DMaaPErrorMessages errorMessages;
-
+
/**
* User defined exception class for kafka consumer cache
- *
+ *
* @author nilanjana.maity
*
*/
public class KafkaConsumerCacheException extends Exception {
/**
* To throw the exception
- *
+ *
* @param t
*/
KafkaConsumerCacheException(Throwable t) {
}
/**
- *
+ *
* @param s
*/
public KafkaConsumerCacheException(String s) {
private static final long serialVersionUID = 1L;
}
+ private enum Status {
+ NOT_STARTED, CONNECTED, DISCONNECTED, SUSPENDED
+ }
+
/**
* Creates a KafkaConsumerCache object. Before it is used, you must call
* startCache()
/**
* Start the cache service. This must be called before any get/put
* operations.
- *
+ *
* @param mode
* DMAAP or cambria
* @param curator
EnsurePath ensurePath = new EnsurePath(fBaseZkPath);
ensurePath.ensure(curator.getZookeeperClient());
-
-
+
+
long freq = kDefault_SweepEverySeconds;
String strkSetting_SweepEverySeconds = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
kSetting_SweepEverySeconds);
/**
* Getting the curator oject to start the zookeeper connection estabished
- *
+ *
* @param curator
* @return curator object
*/
* valid) In addition, this method waits for all other consumer caches in
* the cluster to release their ownership and delete their version of this
* consumer.
- *
+ *
* @param topic
* @param consumerGroupId
* @param clientId
return true;
}
- // private final rrNvReadable fSettings;
- private MetricsSet fMetrics;
- private final String fBaseZkPath;
- private final ScheduledExecutorService fSweepScheduler;
- private String fApiId;
-
public void setfMetrics(final MetricsSet metrics) {
this.fMetrics = metrics;
}
this.fApiId = id;
}
- private final ConnectionStateListener listener;
-
- private ConcurrentHashMap<String, Kafka011Consumer> fConsumers;
- private PathChildrenCache curatorConsumerCache;
-
- private volatile Status status;
-
private void handleReconnection() {
log.info("Reading current cache data from ZK and synchronizing local cache");
}
public void sweep() {
- final LinkedList<String> removals = new LinkedList<String>();
+ final LinkedList<String> removals = new LinkedList<>();
long mustTouchEveryMs = kDefault_MustTouchEveryMs;
String strkSetting_TouchEveryMs = AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop,
kSetting_TouchEveryMs);
- if (null != strkSetting_TouchEveryMs) {
+ if (null != strkSetting_TouchEveryMs && !strkSetting_TouchEveryMs.isEmpty()) {
mustTouchEveryMs = Long.parseLong(strkSetting_TouchEveryMs);
}
-
+
final long oldestAllowedTouchMs = System.currentTimeMillis() - mustTouchEveryMs;
for (Entry<String, Kafka011Consumer> e : fConsumers.entrySet()) {
/**
* Creating a thread to run the sweep method
- *
+ *
* @author nilanjana.maity
*
*/
/**
* This method is to drop consumer
- *
+ *
* @param topic
* @param consumerGroup
* @param clientId
}
private static final EELFLogger log = EELFManager.getInstance().getLogger(KafkaConsumerCache.class);
-
+
}
\ No newline at end of file
* 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=========================================================
- *
+ *
* ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
+ *
*******************************************************************************/
package org.onap.dmaap.filemonitor;
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
-import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
-
import javax.annotation.PostConstruct;
-import com.att.eelf.configuration.EELFLogger;
-import com.att.eelf.configuration.EELFManager;
-
-
-//import com.att.ssf.filemonitor.FileChangedListener;
-//import com.att.ssf.filemonitor.FileMonitor;
-
/**
* ServicePropertyService class
* @author rajashree.khare
try {
getFileList(FILE_CHANGE_LISTENER_LOC);
- /*for (File file : fileList) {
- FileChangedListener fileChangedListener = this.fileChangedListener;
- Object filePropertiesMap = this.filePropertiesMap;
- Method m = filePropertiesMap.getClass().getMethod(
- "refresh", File.class);
- m.invoke(filePropertiesMap, file);
- FileMonitor fm = FileMonitor.getInstance();
- fm.addFileChangedListener(file, fileChangedListener,
- loadOnStartup);
-
- }*/
} catch (Exception ex) {
logger.error("Error creating property map ", ex);
}
private void getFileList(String dirName) throws IOException {
File directory = new File(dirName);
- FileInputStream fis = null;
if (fileList == null)
- fileList = new ArrayList<File>();
+ fileList = new ArrayList<>();
// get all the files that are ".json" or ".properties", from a directory
// & it's sub-directories
for (File file : fList) {
// read service property files from the configuration file
if (file.isFile() && file.getPath().endsWith(USER_CONFIG_FILE)) {
- try {
- fis = new FileInputStream(file);
+ try(FileInputStream fis = new FileInputStream(file)) {
+
Properties prop = new Properties();
prop.load(fis);
}
} catch (Exception ioe) {
logger.error("Error reading the file stream ", ioe);
- } finally {
- fis.close();
}
} else if (file.isDirectory()) {
getFileList(file.getPath());
* 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.
import org.apache.log4j.Logger;
@RunWith(Suite.class)
-@SuiteClasses({ ServicePropertiesListenerTest.class, ServicePropertiesMapTest.class,
+@SuiteClasses({ServicePropertiesMapTest.class,
ServicePropertyServiceTest.class, })
public class JUnitTestSuite {
private static final Logger LOGGER = Logger.getLogger(JUnitTestSuite.class);
public static void main(String[] args) {
LOGGER.info("Running the test suite");
-
+
TestSuite tstSuite = new TestSuite();
LOGGER.info("Total Test Counts " + tstSuite.countTestCases());
}
+++ /dev/null
-/*-
- * ============LICENSE_START=======================================================
- * ONAP Policy Engine
- * ================================================================================
- * Copyright (C) 2017 AT&T Intellectual Property. 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.dmaap.filemonitor;
-
-import static org.junit.Assert.*;
-
-import java.io.File;
-
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-public class ServicePropertiesListenerTest {
-
- @Before
- public void setUp() throws Exception {
- }
-
- @After
- public void tearDown() throws Exception {
- }
-
- //@Test
- public void testUpdate() {/*
-
- ServicePropertiesListener listener = new ServicePropertiesListener();
-
- try {
- listener.update(new File(":/file"));
- } catch (Exception e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
-
- assertTrue(true);
-
- */}
-
-}
\ No newline at end of file
+++ /dev/null
-/*******************************************************************************
- * ============LICENSE_START=======================================================
- * org.onap.dmaap
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. 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=========================================================
- *
- * ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
- *******************************************************************************/
-package org.onap.dmaap.mr.cambria;
-
-import junit.framework.TestCase;
-import org.junit.Ignore;
-
-import java.util.HashMap;
-import java.util.Map;
-
-@Ignore
-public class CambriaApiTestCase extends TestCase {
-
- @Override
- protected void setUp() throws Exception {
- final Map<String, String> argMap = new HashMap<String, String> ();
-
- argMap.put("broker.type", "memory");
- argMap.put("accounts.dao.class", "com.att.nsa.fe3c.dao.memory.MemoryDAOFactory");
- argMap.put("topic.dao.class", "com.att.nsa.fe3c.dao.memory.MemoryDAOFactory");
-
- //CambriaApiServer.start(argMap);
- System.out.println("setUp() complete");
- }
-
- public void tearDown() throws Exception {
- System.out.println("tearDown() started");
- //CambriaApiServer.stop();
- System.out.println("tearDown() complete");
- }
-}
+++ /dev/null
-/*******************************************************************************
- * ============LICENSE_START=======================================================
- * org.onap.dmaap
- * ================================================================================
- * Copyright © 2017 AT&T Intellectual Property. 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=========================================================
- *
- * ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
- *******************************************************************************/
-package org.onap.dmaap.mr.cambria;
-
-import org.junit.Test;
-
-public class CambriaRateLimiterTest
-{
- @Test
- public void testRateLimiter ()
- {
- /*final NsaTestClock clock = new NsaTestClock(1, false);
-
- final String topic = "topic";
- final String consumerGroup = "group";
- final String clientId = "id";
-
- final int window = 5;
-
- // rate limit: 1 empty call/min avg over 5 minutes, with 10ms delay
- final CambriaRateLimiter rater = new CambriaRateLimiter ( 1.0, window, 10 );
- try
- {
- // prime with a call to start rate window
- rater.onCall ( topic, consumerGroup, clientId );
- rater.onSend ( topic, consumerGroup, clientId, 1 );
- clock.addMs ( 1000*60*window );
-
- // rate should now be 0, with a good window
- for ( int i=0; i<4; i++ )
- {
- clock.addMs ( 1000*15 );
- rater.onCall ( topic, consumerGroup, clientId );
- rater.onSend ( topic, consumerGroup, clientId, 0 );
- }
- // rate is now 0.8 = 4 calls in last 5 minutes = 4/5 = 0.8
-
- clock.addMs ( 1000*15 );
- rater.onCall ( topic, consumerGroup, clientId );
- rater.onSend ( topic, consumerGroup, clientId, 0 );
- // rate = 1.0 = 5 calls in last 5 mins
-
- clock.addMs ( 1000 );
- rater.onCall ( topic, consumerGroup, clientId );
- rater.onSend ( topic, consumerGroup, clientId, 0 );
- // rate = 1.2 = 6 calls in last 5 mins, should fire
-
- fail ( "Should have thrown rate limit exception." );
- }
- catch ( CambriaApiException x )
- {
- // good
- }*/
- }
-}
import org.junit.runners.Suite.SuiteClasses;
@RunWith(Suite.class)
-@SuiteClasses({CambriaApiExceptionTest.class, CambriaApiVersionInfoTest.class,
- CambriaApiTestCase.class, CambriaRateLimiterTest.class,})
+@SuiteClasses({CambriaApiExceptionTest.class, CambriaApiVersionInfoTest.class})
public class JUnitTestSuite {
private static final Logger LOGGER = LogManager.getLogger(JUnitTestSuite.class);
* 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.
+ * limitations under the License.#
* ============LICENSE_END=========================================================
- *
+ *
* ECOMP is a trademark and service mark of AT&T Intellectual Property.
- *
+ *
*******************************************************************************/
package org.onap.dmaap.mr.cambria.backends.kafka;
import static org.junit.Assert.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import com.att.ajsc.filemonitor.AJSCPropertiesMap;
import org.apache.kafka.clients.consumer.KafkaConsumer;
@RunWith(PowerMockRunner.class)
@PrepareForTest({ AJSCPropertiesMap.class })
public class Kafka011ConsumerTest {
-
-
+
+
@Mock
private KafkaConsumer<String, String> cc;
@Mock
@Before
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
-
+
}
@After
public void tearDown() throws Exception {
}
-
+
@Test
public void testKafka011Consumer() {
PowerMockito.mockStatic(AJSCPropertiesMap.class);
consumer.touch();
consumer.setOffset(10);
} catch (Exception e) {
-
+
}
assertNotNull(consumer);
assertNotNull(consumer.getConsumer());
assertNotNull(consumer.getConsumerGroup());
assertNotNull(consumer.getConsumerId());
assertNotNull(consumer.getConsumerId());
- assertNotNull(consumer.getCreateTimeMs());
- assertNotNull(consumer.getLastAccessMs());
+ assertTrue(consumer.getCreateTimeMs() > 0);
+ assertTrue(consumer.getLastAccessMs() > 0);
assertNotNull(consumer.getName());
- assertNotNull(consumer.getOffset());
- assertNotNull(consumer.getLastTouch());
-
-
+ assertTrue(consumer.getOffset() > 0);
+ assertTrue(consumer.getLastTouch() > 0);
+
+
}
-
+
}
* 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.
package org.onap.dmaap.mr.cambria.backends.kafka;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertTrue;
+import com.att.ajsc.filemonitor.AJSCPropertiesMap;
+import java.util.Collection;
import java.util.concurrent.ConcurrentHashMap;
-
import org.apache.curator.framework.CuratorFramework;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
+import org.mockito.InjectMocks;
import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.modules.junit4.PowerMockRunner;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-
-import com.att.ajsc.filemonitor.AJSCPropertiesMap;
-
+import org.mockito.junit.MockitoJUnitRunner;
+import org.onap.dmaap.dmf.mr.backends.Consumer;
import org.onap.dmaap.dmf.mr.backends.MetricsSet;
import org.onap.dmaap.dmf.mr.backends.kafka.Kafka011Consumer;
import org.onap.dmaap.dmf.mr.backends.kafka.KafkaConsumerCache;
import org.onap.dmaap.dmf.mr.backends.kafka.KafkaConsumerCache.KafkaConsumerCacheException;
-import org.onap.dmaap.dmf.mr.constants.CambriaConstants;
-@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"})
-@RunWith(PowerMockRunner.class)
-@PrepareForTest({ AJSCPropertiesMap.class })
+
+@RunWith(MockitoJUnitRunner.class)
public class KafkaConsumerCacheTest {
- private KafkaConsumerCache kafkaConsumerCache =null;
+
@Mock
private ConcurrentHashMap<String, Kafka011Consumer> fConsumers;
@Mock
private MetricsSet fMetrics;
+ @Mock
+ private AJSCPropertiesMap ajscPropertiesMap;
+ @InjectMocks
+ private KafkaConsumerCache kafkaConsumerCache;
@Before
public void setUp() throws Exception {
- MockitoAnnotations.initMocks(this);
-
}
@After
public void tearDown() throws Exception {
}
-
@Test
public void testSweep() {
- kafkaConsumerCache = new KafkaConsumerCache();
- PowerMockito.mockStatic(AJSCPropertiesMap.class);
- PowerMockito.when(AJSCPropertiesMap.getProperty(CambriaConstants.msgRtr_prop, "kSetting_TouchEveryMs")).thenReturn("100");
kafkaConsumerCache.sweep();
-
+ Collection<? extends Consumer> cachedConsumers =
+ kafkaConsumerCache.getConsumers();
+ assertTrue(cachedConsumers.isEmpty());
}
-
+
// DOES NOT WORK
@Test
/*
* KafkaConsumerCache kafka = null;
- *
+ *
* try { kafka = new KafkaConsumerCache("123", null);
- *
+ *
* } catch (NoClassDefFoundError e) { try { kafka.startCache("DMAAP",
* null); } catch (NullPointerException e1) { // TODO Auto-generated
* catch block assertTrue(true); } catch (KafkaConsumerCacheException
* e1) { // TODO Auto-generated catch block e1.printStackTrace(); } }
*/
-
+
new CuratorFrameworkImpl();
new MetricsSetImpl();
KafkaConsumerCache kafka=null;
/*
* @Test public void testStopCache() {
- *
+ *
* KafkaConsumerCache kafka = null; new CuratorFrameworkImpl(); new
* MetricsSetImpl(); try { kafka = new KafkaConsumerCache("123", null);
* kafka.stopCache(); } catch (NoClassDefFoundError e) {
- *
+ *
* }
- *
+ *
* }
*/
+ @Test(expected= KafkaConsumerCacheException.class)
+ public void testGetConsumerThrowNoConnectionToCache() throws KafkaConsumerCacheException{
+ KafkaConsumerCache kafka = new KafkaConsumerCache();
+ kafka.getConsumerFor("testTopic", "CG1", "23");
+ }
+
@Test
public void testGetConsumerFor() {
}
}
- }
+ }
@Test
public void testSignalOwnership() {
}
} catch (NoClassDefFoundError e) {}
- //
+ //
}
@Test
}
}
-
+
}
Code Review / dmaap / messagerouter / messageservice.git / commitdiff