target
.metadata/
/bin/
+bin/
derby.log
.checkstyle
testBadItem001.setByteValue(BYTE_VAL);
testBadItem002.setIntValue(INT_VAL);
testBadItem003.setLongValue(LONG_VAL);
- testBadItem004.setFloatValue(new Float(FLOAT_VAL));
+ testBadItem004.setFloatValue(FLOAT_VAL);
testBadItem005.setDoubleValue(PI_VAL);
testBadItem006.setStringValue(STRING_GLOBAL_VAL);
testBadItem007.setLongValue(testDate.getTime());
package org.onap.policy.apex.context.test.locking;
+import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import org.onap.policy.apex.context.ContextAlbum;
import org.onap.policy.apex.context.ContextException;
import org.onap.policy.apex.context.impl.distribution.DistributorFactory;
import org.onap.policy.apex.context.test.concepts.TestContextLongItem;
import org.onap.policy.apex.context.test.factory.TestContextAlbumFactory;
+import org.onap.policy.apex.context.test.utils.IntegrationThreadFactory;
import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
import org.onap.policy.apex.model.basicmodel.handling.ApexModelException;
* @author Liam Fallon (liam.fallon@ericsson.com)
*/
public class ConcurrentContext {
- private static final int TEN_MILLISECONDS = 10;
-
// Logger for this class
private static final XLogger LOGGER = XLoggerFactory.getXLogger(ConcurrentContext.class);
LOGGER.debug("starting JVMs and threads . . .");
- final Thread[] threadArray = new Thread[threadCount];
+ final String name = getThreadFactoryName(jvmCount, testType);
+ final IntegrationThreadFactory threadFactory = new IntegrationThreadFactory(name);
+ final ExecutorService executorService = Executors.newFixedThreadPool(threadCount, threadFactory);
+
+ final List<Closeable> tasks = new ArrayList<>(threadCount);
+
+ addShutDownHook(tasks);
// Check if we have a single JVM or multiple JVMs
- int runningThreadCount = -1;
if (jvmCount == 1) {
// Run everything in this JVM
for (int t = 0; t < threadCount; t++) {
- threadArray[t] = new Thread(new ConcurrentContextThread(0, t, threadLoops));
- threadArray[t].setName(testType + ":TestConcurrentContextThread_0_" + t);
- threadArray[t].start();
+ final ConcurrentContextThread task = new ConcurrentContextThread(0, t, threadLoops);
+ tasks.add(task);
+ executorService.execute(task);
}
- runningThreadCount = threadCount;
} else {
// Spawn JVMs to run the tests
for (int j = 0; j < jvmCount; j++) {
- threadArray[j] = new Thread(new ConcurrentContextJVMThread(testType, j, threadCount, threadLoops));
- threadArray[j].setName(testType + ":TestConcurrentContextJVMThread_" + j);
- threadArray[j].start();
+ final ConcurrentContextJVMThread task =
+ new ConcurrentContextJVMThread(testType, j, threadCount, threadLoops);
+ tasks.add(task);
+ executorService.execute(task);
}
- runningThreadCount = jvmCount;
}
- boolean allFinished;
- do {
- allFinished = true;
- for (int i = 0; i < runningThreadCount; i++) {
- if (threadArray[i].isAlive()) {
- allFinished = false;
+ try {
+ executorService.shutdown();
+ // wait for threads to finish, if not Timeout
+ executorService.awaitTermination(10, TimeUnit.MINUTES);
+ } catch (final InterruptedException interruptedException) {
+ LOGGER.error("Exception while waiting for threads to finish", interruptedException);
+ }
+
+ LOGGER.info("Shutting down now ...");
+ executorService.shutdownNow();
+
+ return concurrentContext.verifyAndClearContext(jvmCount, threadCount, threadLoops);
+ }
+
+
+ private void addShutDownHook(final List<Closeable> tasks) {
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ LOGGER.info("Shutting down ...");
+ for (final Closeable task : tasks) {
try {
- Thread.sleep(TEN_MILLISECONDS);
- } catch (final Exception e) {
+ task.close();
+ } catch (final IOException ioException) {
+ LOGGER.error("Unable to close task ... ", ioException);
}
- break;
}
}
- } while (!allFinished);
+ });
+ }
- return concurrentContext.verifyAndClearContext(jvmCount, threadCount, threadLoops);
+
+ private String getThreadFactoryName(final int jvmCount, final String testType) {
+ return jvmCount == 1 ? testType + ":TestConcurrentContextThread_0_"
+ : testType + ":TestConcurrentContextJVMThread_";
}
/**
package org.onap.policy.apex.context.test.locking;
-import com.google.gson.Gson;
-
import java.net.InetAddress;
import java.net.NetworkInterface;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
+import java.util.List;
import java.util.Map.Entry;
import java.util.TreeSet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import org.onap.policy.apex.context.ContextAlbum;
import org.onap.policy.apex.context.Distributor;
import org.onap.policy.apex.context.impl.distribution.DistributorFactory;
import org.onap.policy.apex.context.test.factory.TestContextAlbumFactory;
+import org.onap.policy.apex.context.test.utils.IntegrationThreadFactory;
import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
import org.onap.policy.apex.model.basicmodel.concepts.AxArtifactKey;
import org.onap.policy.apex.model.basicmodel.service.AbstractParameters;
import org.slf4j.ext.XLogger;
import org.slf4j.ext.XLoggerFactory;
+import com.google.gson.Gson;
+
/**
* The Class ConcurrentContextJVM tests concurrent use of context in a single JVM.
*
// Logger for this class
private static final XLogger LOGGER = XLoggerFactory.getXLogger(ConcurrentContextJVM.class);
- private static final int TEN_MILLISECONDS = 10;
private static final int IPV4_ADDRESS_LENGTH = 4;
+ private final int jvmNo;
+
+ private final int threadCount;
+
+ private final int threadLoops;
+
+ private final ExecutorService executorService;
+
/**
* The Constructor.
*
* @param threadLoops the thread loops
* @throws ApexException the apex exception
*/
- private ConcurrentContextJVM(final String testType, final int jvmNo, final int threadCount, final int threadLoops)
- throws ApexException {
- super();
+ private ConcurrentContextJVM(final String testType, final int jvmNo, final int threadCount, final int threadLoops) {
+ this.jvmNo = jvmNo;
+ this.threadCount = threadCount;
+ this.threadLoops = threadLoops;
+ final String name = testType + ":ConcurrentContextThread_" + jvmNo;
+ this.executorService = Executors.newFixedThreadPool(threadCount, new IntegrationThreadFactory(name));
+ }
+
+ public void execute() throws ApexException {
LOGGER.debug("starting JVMs and threads . . .");
final AxArtifactKey distributorKey = new AxArtifactKey("ApexDistributor" + jvmNo, "0.0.1");
assert (lTypeAlbum != null);
lTypeAlbum.setUserArtifactStack(usedArtifactStackArray);
- final Thread[] threadArray = new Thread[threadCount];
+ final List<Future<?>> tasks = new ArrayList<>(threadCount);
for (int t = 0; t < threadCount; t++) {
- threadArray[t] = new Thread(new ConcurrentContextThread(jvmNo, t, threadLoops));
- threadArray[t].setName(testType + ":ConcurrentContextThread_" + jvmNo + "_" + t);
- threadArray[t].start();
- LOGGER.debug("started thread " + threadArray[t].getName());
+ tasks.add(executorService.submit(new ConcurrentContextThread(jvmNo, t, threadLoops)));
}
- boolean allFinished;
- do {
- allFinished = true;
- for (int t = 0; t < threadCount; t++) {
- if (threadArray[t].isAlive()) {
- allFinished = false;
- try {
- Thread.sleep(TEN_MILLISECONDS);
- } catch (final Exception e) {
- }
- break;
- }
- }
- } while (!allFinished);
+ try {
+ executorService.shutdown();
+ // wait for threads to finish, if not Timeout
+ executorService.awaitTermination(10, TimeUnit.MINUTES);
+ } catch (final InterruptedException interruptedException) {
+ LOGGER.error("Exception while waiting for threads to finish", interruptedException);
+ }
LOGGER.debug("threads finished, end value is {}", lTypeAlbum.get("testValue"));
contextDistributor.clear();
+ LOGGER.info("Shutting down now ... ");
+ executorService.shutdownNow();
}
+
+
/**
* The main method.
*
// CHECKSTYLE:OFF: checkstyle:magicNumber
// An even number of arguments greater than 3
- if (args.length < 4 || (args.length % 2 != 0)) {
+ if (args.length < 7) {
LOGGER.error("invalid arguments: " + Arrays.toString(args));
LOGGER.error(
"usage: TestConcurrentContextJVM testType jvmNo threadCount threadLoops [parameterKey parameterJson].... ");
int jvmNo = -1;
int threadCount = -1;
int threadLoops = -1;
+ String hazelCastfileLocation = null;
try {
jvmNo = Integer.parseInt(args[1]);
return;
}
- for (int p = 4; p < args.length - 1; p += 2) {
+ try {
+ hazelCastfileLocation = args[4].trim();
+ } catch (final Exception e) {
+ LOGGER.error("invalid argument hazelcast file location", e);
+ return;
+ }
+
+ System.setProperty("hazelcast.config", hazelCastfileLocation);
+
+ for (int p = 5; p < args.length - 1; p += 2) {
@SuppressWarnings("rawtypes")
final Class parametersClass = Class.forName(args[p]);
final AbstractParameters parameters =
}
try {
- new ConcurrentContextJVM(args[0], jvmNo, threadCount, threadLoops);
+ final ConcurrentContextJVM concurrentContextJVM =
+ new ConcurrentContextJVM(args[0], jvmNo, threadCount, threadLoops);
+ concurrentContextJVM.execute();
+
} catch (final Exception e) {
LOGGER.error("error running test in JVM", e);
return;
}
// CHECKSTYLE:ON: checkstyle:magicNumber
+
+
}
/**
*/
public static void configure() throws Exception {
System.setProperty("java.net.preferIPv4Stack", "true");
- System.setProperty("hazelcast.config", "src/test/resources/hazelcast/hazelcast.xml");
-
- // The JGroups IP address must be set to a real (not loopback) IP address for Infinispan to work. IN order to
+ // The JGroups IP address must be set to a real (not loopback) IP address for Infinispan to
+ // work. IN order to
// ensure that all
- // the JVMs in a test pick up the same IP address, this function sets te address to be the first non-loopback
+ // the JVMs in a test pick up the same IP address, this function sets te address to be the
+ // first non-loopback
// IPv4 address
// on a host
final TreeSet<String> ipAddressSet = new TreeSet<String>();
package org.onap.policy.apex.context.test.locking;
-import com.google.gson.Gson;
-
import java.io.BufferedReader;
+import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import org.slf4j.ext.XLogger;
import org.slf4j.ext.XLoggerFactory;
+import com.google.gson.Gson;
+
/**
* The Class TestConcurrentContextThread tests concurrent use of context.
*
* @author Liam Fallon (liam.fallon@ericsson.com)
*/
-public class ConcurrentContextJVMThread implements Runnable {
+public class ConcurrentContextJVMThread implements Runnable, Closeable {
// Logger for this class
private static final XLogger LOGGER = XLoggerFactory.getXLogger(ConcurrentContextJVMThread.class);
private final int jvm;
private final int threadCount;
private final int target;
+ private Process process = null;
/**
* The Constructor.
final List<String> commandList = new ArrayList<>();
commandList.add(System.getProperty("java.home") + System.getProperty("file.separator") + "bin"
+ System.getProperty("file.separator") + "java");
+ commandList.add("-Xms512m");
+ commandList.add("-Xmx512m");
commandList.add("-cp");
commandList.add(System.getProperty("java.class.path"));
commandList.add(ConcurrentContextJVM.class.getCanonicalName());
commandList.add(new Integer(jvm).toString());
commandList.add(new Integer(threadCount).toString());
commandList.add(new Integer(target).toString());
+ commandList.add(System.getProperty("hazelcast.config"));
for (final Entry<Class<?>, AbstractParameters> parameterServiceEntry : ParameterService.getAll()) {
commandList.add(parameterServiceEntry.getKey().getCanonicalName());
// Run the JVM
final ProcessBuilder processBuilder = new ProcessBuilder(commandList);
processBuilder.redirectErrorStream(true);
- Process process;
+
try {
process = processBuilder.start();
LOGGER.error("Error occured while writing JVM Output for command ", ioException);
}
}
+
+
+ @Override
+ public void close() {
+ LOGGER.info("Shutting down {} thread ...", Thread.currentThread().getName());
+ if (process != null) {
+ LOGGER.info("Destroying process ...");
+ process.destroy();
+ }
+ }
}
package org.onap.policy.apex.context.test.locking;
+import java.io.Closeable;
+
import org.onap.policy.apex.context.ContextAlbum;
import org.onap.policy.apex.context.ContextException;
import org.onap.policy.apex.context.Distributor;
*
* @author Liam Fallon (liam.fallon@ericsson.com)
*/
-public class ConcurrentContextThread implements Runnable {
+public class ConcurrentContextThread implements Runnable, Closeable {
+ private static final String VALUE = "testValue";
// Logger for this class
private static final XLogger LOGGER = XLoggerFactory.getXLogger(ConcurrentContextThread.class);
private final Distributor distributor;
}
try {
- lTypeAlbum.lockForWriting("testValue");
- final TestContextLongItem item = (TestContextLongItem) lTypeAlbum.get("testValue");
+ lTypeAlbum.lockForWriting(VALUE);
+ final TestContextLongItem item = (TestContextLongItem) lTypeAlbum.get(VALUE);
final long value = item.getLongValue();
LOGGER.info("completed TestConcurrentContextThread_" + jvm + "_" + instance + ", value=" + value);
} catch (final Exception e) {
LOGGER.error("failed TestConcurrentContextThread_" + jvm + "_" + instance);
} finally {
try {
- lTypeAlbum.unlockForWriting("testValue");
+ lTypeAlbum.unlockForWriting(VALUE);
distributor.shutdown();
} catch (final ContextException e) {
LOGGER.error("could not unlock test context album item", e);
private void updateAlbum(final ContextAlbum lTypeAlbum) throws Exception {
for (int i = 0; i < threadLoops; i++) {
try {
- lTypeAlbum.lockForWriting("testValue");
- TestContextLongItem item = (TestContextLongItem) lTypeAlbum.get("testValue");
+ lTypeAlbum.lockForWriting(VALUE);
+ TestContextLongItem item = (TestContextLongItem) lTypeAlbum.get(VALUE);
if (item != null) {
long value = item.getLongValue();
item.setLongValue(++value);
} else {
item = new TestContextLongItem(0L);
}
- lTypeAlbum.put("testValue", item);
+ lTypeAlbum.put(VALUE, item);
} finally {
- lTypeAlbum.unlockForWriting("testValue");
+ lTypeAlbum.unlockForWriting(VALUE);
}
}
}
+
+ @Override
+ public void close() {
+ LOGGER.info("Shutting down {} thread ...", Thread.currentThread().getName());
+ }
}
public static final String TEST_POLICY_CONTEXT_ITEM = TestPolicyContextItem.class.getName();
public static final TimeZone TIME_ZONE = TimeZone.getTimeZone("Europe/Dublin");
- public static final AxArtifactKey[] USED_ARTIFACT_STACK_ARRAY = {new AxArtifactKey("testC-top", VERSION),
+ public static AxArtifactKey[] USED_ARTIFACT_STACK_ARRAY = {new AxArtifactKey("testC-top", VERSION),
new AxArtifactKey("testC-next", VERSION), new AxArtifactKey("testC-bot", VERSION)};
private Constants() {}
--- /dev/null
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2018 Ericsson. 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+package org.onap.policy.apex.context.test.utils;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.ext.XLogger;
+import org.slf4j.ext.XLoggerFactory;
+
+public class IntegrationThreadFactory implements ThreadFactory {
+ private static final XLogger LOGGER = XLoggerFactory.getXLogger(IntegrationThreadFactory.class);
+
+
+ private final String threadFactoryName;
+
+ private final AtomicInteger counter = new AtomicInteger();
+
+ public IntegrationThreadFactory(final String threadFactoryName) {
+ this.threadFactoryName = threadFactoryName;
+ }
+
+ @Override
+ public Thread newThread(final Runnable runnable) {
+ final Thread thread = new Thread(runnable);
+ thread.setName(threadFactoryName + "_" + counter.getAndIncrement());
+ LOGGER.debug("started thread " + thread.getName());
+ return thread;
+ }
+
+}
</dependencySets>
<fileSets>
<fileSet>
- <directory>${project.basedir}/src/main/package/bin</directory>
+ <directory>${project.basedir}/src/main/package/scripts</directory>
<includes>
<include>*</include>
</includes>
import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
import org.onap.policy.apex.model.basicmodel.handling.ApexModelException;
import org.onap.policy.apex.model.basicmodel.service.ParameterService;
+import org.onap.policy.apex.model.utilities.ResourceUtils;
import org.onap.policy.apex.plugins.context.distribution.hazelcast.HazelcastContextDistributor;
import org.onap.policy.apex.plugins.context.distribution.infinispan.InfinispanContextDistributor;
import org.onap.policy.apex.plugins.context.distribution.infinispan.InfinispanDistributorParameters;
* @author Liam Fallon (liam.fallon@ericsson.com)
*/
public class TestConcurrentContext {
+ private static final String HAZELCAST_CONFIG = "hazelcast.config";
+
+ private static final String JAVA_NET_PREFER_IPV4_STACK = "java.net.preferIPv4Stack";
+ private static final String HAZELCAST_XML_FILE = "src/test/resources/hazelcast/hazelcast.xml";
+
// Logger for this class
private static final XLogger logger = XLoggerFactory.getXLogger(TestConcurrentContext.class);
@BeforeClass
public static void configure() throws Exception {
- System.setProperty("java.net.preferIPv4Stack", "true");
- System.setProperty("hazelcast.config", "src/test/resources/hazelcast/hazelcast.xml");
+ System.setProperty(JAVA_NET_PREFER_IPV4_STACK, "true");
+ final String hazelCastfileLocation = ResourceUtils.getFilePath4Resource(HAZELCAST_XML_FILE);
+ System.setProperty(HAZELCAST_CONFIG, hazelCastfileLocation);
// The JGroups IP address must be set to a real (not loopback) IP address for Infinispan to
// work. IN order to ensure that all
</interfaces>
</network>
<properties>
- <property name="hazelcast.icmp.enabled">true</property>
+ <property name="hazelcast.icmp.enabled">false</property>
<property name="hazelcast.logging.type">slf4j</property>
<!-- disable the hazelcast shutdown hook - prefer to control the shutdown
in code -->
<property name="hazelcast.shutdownhook.enabled">false</property>
<property name="hazelcast.graceful.shutdown.max.wait">60</property>
+ <property name="hazelcast.operation.generic.thread.count">1</property>
+ <property name="hazelcast.operation.thread.count">1</property>
+ <property name="hazelcast.clientengine.thread.count">1</property>
+ <property name="hazelcast.client.event.thread.count">1</property>
+ <property name="hazelcast.event.thread.count">1</property>
+ <property name="hazelcast.io.output.thread.count">1</property>
+ <property name="hazelcast.io.thread.count">1</property>
+ <property name="hazelcast.executor.client.thread.count">1</property>
+ <property name="hazelcast.clientengine.thread.count">1</property>
</properties>
<executor-service>
- <pool-size>16</pool-size>
+ <pool-size>2</pool-size>
</executor-service>
</hazelcast>
+++ /dev/null
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ============LICENSE_START=======================================================
- Copyright (C) 2016-2018 Ericsson. 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.
-
- SPDX-License-Identifier: Apache-2.0
- ============LICENSE_END=========================================================
--->
-
-<configuration>
- <contextName>Apex</contextName>
- <statusListener class="ch.qos.logback.core.status.OnConsoleStatusListener" />
- <property name="LOG_DIR" value="${java.io.tmpdir}/apex_logging/" />
-
- <!-- USE FOR STD OUT ONLY -->
- <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
- <encoder>
- <Pattern>%d %contextName [%t] %level %logger{36} - %msg%n</Pattern>
- </encoder>
- </appender>
-
- <logger name="org.apache.zookeeper.ClientCnxn" level="OFF" additivity="false">
- <appender-ref ref="STDOUT" />
- </logger>
-
- <appender name="FILE"
- class="ch.qos.logback.core.FileAppender">
- <file>${LOG_DIR}/apex.log</file>
- <encoder>
- <pattern>%d %-5relative [procId=${processId}] [%thread] %-5level
- %logger{26} - %msg %n %ex{full}</pattern>
- </encoder>
- </appender>
-
- <appender name="CTXT_FILE"
- class="ch.qos.logback.core.FileAppender">
- <file>${LOG_DIR}/apex_ctxt.log</file>
- <encoder>
- <pattern>%d %-5relative [procId=${processId}] [%thread] %-5level
- %logger{26} - %msg %n %ex{full}</pattern>
- </encoder>
- </appender>
-
- <logger name="org.infinispan" level="info" additivity="false">
- <appender-ref ref="STDOUT" />
- </logger>
-
- <logger name="com.ericsson.apex.context.monitoring" level="TRACE" additivity="false">
- <appender-ref ref="STDOUT" />
- </logger>
-
- <logger name="com.ericsson.apex.context.monitoring" level="TRACE" additivity="false">
- <appender-ref ref="CTXT_FILE" />
- </logger>
-
- <logger name="org.jgroups" level="off" additivity="false">
- <appender-ref ref="STDOUT" />
- </logger>
-
- <root level="info">
- <appender-ref ref="STDOUT" />
- </root>
-
-</configuration>
@AfterClass
public static void shutdownDummyKafkaServer() throws IOException {
- kafkaServer.shutdown();
- zkClient.close();
- zkServer.shutdown();
+ if (kafkaServer != null) {
+ kafkaServer.shutdown();
+ }
+ if (zkClient != null) {
+ zkClient.close();
+ }
+ if (zkServer != null) {
+ zkServer.shutdown();
+ }
}
@Test
Code Review / policy / apex-pdp.git / commitdiff