<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.projectlombok</groupId>
+ <artifactId>lombok</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
<dependency>
<groupId>org.onap.policy.drools-pdp</groupId>
<artifactId>api-server-pool-state-control</artifactId>
private volatile State state = null;
// storage for additional data
- private Map<Class<?>, Object> adjuncts = new HashMap<Class<?>, Object>();
+ private Map<Class<?>, Object> adjuncts = new HashMap<>();
+
+ // HTTP query parameters
+ private static final String QP_BUCKET = "bucket";
+ private static final String QP_KEYWORD = "keyword";
+ private static final String QP_DEST = "dest";
+ private static final String QP_TTL = "ttl";
// BACKUP data (only buckets for where we are the owner, or a backup)
return bucket.getOwner() == Server.getThisServer();
}
- /**
- * This method is called to start a 'rebalance' operation in a background
- * thread, but it only does this on the lead server. Being balanced means
- * the following:
- * 1) Each server owns approximately the same number of buckets
- * 2) If any server were to fail, and the designated primaries take over
- * for all of that server's buckets, all remaining servers would still
- * own approximately the same number of buckets.
- * 3) If any two servers were to fail, and the designated primaries were
- * to take over for the failed server's buckets (secondaries would take
- * for buckets where the owner and primary are OOS), all remaining
- * servers would still own approximately the same number of buckets.
- * 4) Each server should have approximately the same number of
- * (primary-backup + secondary-backup) buckets that it is responsible for.
- * 5) The primary backup for each bucket must be on the same site as the
- * owner, and the secondary backup must be on a different site.
- */
- private static void rebalance() {
- if (Leader.getLeader() == Server.getThisServer()) {
- Rebalance rb = new Rebalance();
- synchronized (rebalanceLock) {
- // the most recent 'Rebalance' instance is the only valid one
- rebalance = rb;
- }
-
- new Thread("BUCKET REBALANCER") {
- @Override
- public void run() {
- /*
- * copy bucket and host data,
- * generating a temporary internal table.
- */
- rb.copyData();
-
- /*
- * allocate owners for all buckets without an owner,
- * and rebalance bucket owners, if necessary --
- * this takes card of item #1, above.
- */
- rb.allocateBuckets();
-
- /*
- * make sure that primary backups always have the same site
- * as the owner, and secondary backups always have a different
- * site -- this takes care of #5, above.
- */
- rb.checkSiteValues();
-
- /*
- * adjust primary backup lists to take care of item #2, above
- * (taking #5 into account).
- */
- rb.rebalancePrimaryBackups();
-
- /*
- * allocate secondary backups, and take care of items
- * #3 and #4, above (taking #5 into account).
- */
- rb.rebalanceSecondaryBackups();
-
- try {
- synchronized (rebalanceLock) {
- /*
- * if another 'Rebalance' instance has started in the
- * mean time, don't do the update.
- */
- if (rebalance == rb) {
- /*
- * build a message containing all of the updated bucket
- * information, process it internally in this host
- * (lead server), and send it out to others in the
- * "notify list".
- */
- rb.generateBucketMessage();
- rebalance = null;
- }
- }
- } catch (IOException e) {
- logger.error("Exception in Rebalance.generateBucketMessage",
- e);
- }
- }
- }.start();
- }
- }
-
/**
* Handle an incoming /bucket/update REST message.
*
int tag;
while ((tag = dis.readUnsignedByte()) != END_OF_PARAMETERS_TAG) {
switch (tag) {
- case OWNER_UPDATE: {
+ case OWNER_UPDATE:
// <OWNER_UPDATE> <owner-uuid> -- owner UUID specified
bucketChanges = updateBucketInternalOwnerUpdate(bucket, dis, index);
break;
- }
- case OWNER_NULL: {
+ case OWNER_NULL:
// <OWNER_NULL> -- owner UUID should be set to 'null'
if (bucket.getOwner() != null) {
logger.info("Bucket {} owner: {}->null",
}
}
break;
- }
- case PRIMARY_BACKUP_UPDATE: {
+ case PRIMARY_BACKUP_UPDATE:
// <PRIMARY_BACKUP_UPDATE> <primary-backup-uuid> --
// primary backup UUID specified
Server newPrimaryBackup =
bucket.primaryBackup = newPrimaryBackup;
}
break;
- }
- case PRIMARY_BACKUP_NULL: {
+ case PRIMARY_BACKUP_NULL:
// <PRIMARY_BACKUP_NULL> --
// primary backup should be set to 'null'
if (bucket.primaryBackup != null) {
bucket.primaryBackup = null;
}
break;
- }
- case SECONDARY_BACKUP_UPDATE: {
+ case SECONDARY_BACKUP_UPDATE:
// <SECONDARY_BACKUP_UPDATE> <secondary-backup-uuid> --
// secondary backup UUID specified
Server newSecondaryBackup =
bucket.secondaryBackup = newSecondaryBackup;
}
break;
- }
- case SECONDARY_BACKUP_NULL: {
+ case SECONDARY_BACKUP_NULL:
// <SECONDARY_BACKUP_NULL> --
// secondary backup should be set to 'null'
if (bucket.secondaryBackup != null) {
bucket.secondaryBackup = null;
}
break;
- }
default:
logger.error("Illegal tag: {}", tag);
bucket.state = bucket.new NewOwner(true, oldOwner);
} else {
// new owner has been confirmed
- // orig bucket.state.newOwner();
bucket.state.newOwner();
}
}
* selected bucket has no server assigned -- this should only be a
* transient situation, until 'rebalance' is run.
*/
- out.println("Bucket is " + bucketNumber + ", which has no owner");
+ out.format("Bucket is %d, which has no owner\n", bucketNumber);
} else if (server == Server.getThisServer()) {
/*
* the selected bucket is associated with this particular server --
* no forwarding is needed.
*/
- out.println("Bucket is " + bucketNumber
- + ", which is owned by this server: " + server.getUuid());
+ out.format("Bucket is %d, which is owned by this server: %s\n",
+ bucketNumber, server.getUuid());
} else {
/*
* the selected bucket is assigned to a different server -- forward
* the message.
*/
- out.println("Bucket is " + bucketNumber + ": sending from\n"
- + " " + Server.getThisServer().getUuid() + " to \n"
- + " " + server.getUuid());
+ out.format("Bucket is %d: sending from\n"
+ + " %s to\n"
+ + " %s\n",
+ bucketNumber, Server.getThisServer().getUuid(), server.getUuid());
// do a POST call of /bucket/bucketResponse to the remoote server
Entity<String> entity =
// we need to include the 'bucket' and 'keyword' parameters
// in the POST that we are sending out
return webTarget
- .queryParam("bucket", bucketNumber)
- .queryParam("keyword", keyword);
+ .queryParam(QP_BUCKET, bucketNumber)
+ .queryParam(QP_KEYWORD, keyword);
}
/**
if (response == null) {
out.println("Timed out waiting for a response");
} else {
- out.println("Received response code " + response.getStatus());
- out.println("Entity = " + response.readEntity(String.class));
+ out.format("Received response code %s\nEntity = %s\n",
+ response.getStatus(), response.readEntity(String.class));
}
} catch (InterruptedException e) {
out.println(e);
- throw new IOException(e);
+ Thread.currentThread().interrupt();
}
}
}
Server server;
WebTarget webTarget;
- if ((ttl -= 1) > 0
+ ttl -= 1;
+ if (ttl > 0
&& (server = Server.getServer(dest)) != null
&& (webTarget = server.getWebTarget("bucket/sessionData")) != null) {
logger.info("Forwarding 'bucket/sessionData' to uuid {}",
MediaType.APPLICATION_OCTET_STREAM_TYPE);
Response response =
webTarget
- .queryParam("bucket", bucketNumber)
- .queryParam("dest", dest)
- .queryParam("ttl", String.valueOf(ttl))
+ .queryParam(QP_BUCKET, bucketNumber)
+ .queryParam(QP_DEST, dest)
+ .queryParam(QP_TTL, String.valueOf(ttl))
.request().post(entity);
logger.info("/bucket/sessionData response code = {}",
response.getStatus());
* the 'newInstance' method is unable to create the adjunct)
*/
public <T> T getAdjunct(Class<T> clazz) {
- synchronized (adjuncts) {
- // look up the adjunct in the table
- Object adj = adjuncts.get(clazz);
- if (adj == null) {
- // lookup failed -- create one
- try {
- // create the adjunct (may trigger an exception)
- adj = clazz.newInstance();
-
- // update the table
- adjuncts.put(clazz, adj);
- } catch (Exception e) {
- logger.error("Can't create adjunct of {}", clazz, e);
- }
+ Object adj = adjuncts.computeIfAbsent(clazz, key -> {
+ try {
+ // create the adjunct, if needed
+ return clazz.getDeclaredConstructor().newInstance();
+ } catch (Exception e) {
+ logger.error("Can't create adjunct of {}", clazz, e);
+ return null;
}
- return clazz.cast(adj);
- }
+ });
+ return clazz.cast(adj);
}
/**
*/
public Object putAdjunct(Object adj) {
synchronized (adjuncts) {
- Class clazz = adj.getClass();
+ Class<?> clazz = adj.getClass();
return adjuncts.put(clazz, adj);
}
}
// trigger a rebalance (only happens if we are the lead server)
rebalance();
}
+
+ /**
+ * This method is called to start a 'rebalance' operation in a background
+ * thread, but it only does this on the lead server. Being balanced means
+ * the following:
+ * 1) Each server owns approximately the same number of buckets
+ * 2) If any server were to fail, and the designated primaries take over
+ * for all of that server's buckets, all remaining servers would still
+ * own approximately the same number of buckets.
+ * 3) If any two servers were to fail, and the designated primaries were
+ * to take over for the failed server's buckets (secondaries would take
+ * for buckets where the owner and primary are OOS), all remaining
+ * servers would still own approximately the same number of buckets.
+ * 4) Each server should have approximately the same number of
+ * (primary-backup + secondary-backup) buckets that it is responsible for.
+ * 5) The primary backup for each bucket must be on the same site as the
+ * owner, and the secondary backup must be on a different site.
+ */
+ private void rebalance() {
+ if (Leader.getLeader() == Server.getThisServer()) {
+ Rebalance rb = new Rebalance();
+ synchronized (rebalanceLock) {
+ // the most recent 'Rebalance' instance is the only valid one
+ rebalance = rb;
+ }
+
+ new Thread("BUCKET REBALANCER") {
+ @Override
+ public void run() {
+ /*
+ * copy bucket and host data,
+ * generating a temporary internal table.
+ */
+ rb.copyData();
+
+ /*
+ * allocate owners for all buckets without an owner,
+ * and rebalance bucket owners, if necessary --
+ * this takes card of item #1, above.
+ */
+ rb.allocateBuckets();
+
+ /*
+ * make sure that primary backups always have the same site
+ * as the owner, and secondary backups always have a different
+ * site -- this takes care of #5, above.
+ */
+ rb.checkSiteValues();
+
+ /*
+ * adjust primary backup lists to take care of item #2, above
+ * (taking #5 into account).
+ */
+ rb.rebalancePrimaryBackups();
+
+ /*
+ * allocate secondary backups, and take care of items
+ * #3 and #4, above (taking #5 into account).
+ */
+ rb.rebalanceSecondaryBackups();
+
+ try {
+ synchronized (rebalanceLock) {
+ /*
+ * if another 'Rebalance' instance has started in the
+ * mean time, don't do the update.
+ */
+ if (rebalance == rb) {
+ /*
+ * build a message containing all of the updated bucket
+ * information, process it internally in this host
+ * (lead server), and send it out to others in the
+ * "notify list".
+ */
+ rb.generateBucketMessage();
+ rebalance = null;
+ }
+ }
+ } catch (IOException e) {
+ logger.error("Exception in Rebalance.generateBucketMessage",
+ e);
+ }
+ }
+ }.start();
+ }
+ }
}
/* ============================================================ */
return 0;
};
- FutureTask<Integer> ft = new FutureTask(callable);
+ FutureTask<Integer> ft = new FutureTask<>(callable);
MainLoop.queueWork(ft);
try {
ft.get(60, TimeUnit.SECONDS);
- } catch (InterruptedException | ExecutionException | TimeoutException e) {
+ } catch (InterruptedException e) {
+ logger.error("Interrupted", e);
+ Thread.currentThread().interrupt();
+ return;
+ } catch (ExecutionException | TimeoutException e) {
logger.error("Exception in Rebalance.copyData", e);
return;
}
* 'needBuckets' TreeSet: those with the fewest buckets allocated are
* at the head of the list.
*/
- Comparator<TestServer> bucketCount = new Comparator<TestServer>() {
+ Comparator<TestServer> bucketCount = new Comparator<>() {
@Override
public int compare(TestServer s1, TestServer s2) {
int rval = s1.buckets.size() - s2.buckets.size();
// populate a 'TreeSet' of 'AdjustedTestServer' instances based
// the failure of 'failedServer'
- TreeSet<AdjustedTestServer> adjustedTestServers =
- new TreeSet<AdjustedTestServer>();
+ TreeSet<AdjustedTestServer> adjustedTestServers = new TreeSet<>();
for (TestServer server : testServers.values()) {
if (server == failedServer
|| !Objects.equals(siteSocketAddress,
int size = buckets.size();
if (size != 0) {
// generate a linked list of the bucket data to display
- LinkedList<String> data = new LinkedList<String>();
+ LinkedList<String> data = new LinkedList<>();
StringBuilder sb = new StringBuilder();
int count = 8;
// add the bucket number
sb.append(String.format("%4s", bucket.index));
- if ((count -= 1) <= 0) {
+ count -= 1;
+ if (count <= 0) {
// filled up a row --
// add it to the list, and start a new line
data.add(sb.toString());
// when 'System.currentTimeMillis()' reaches this value, we time out
long endTime;
- // If not 'null', we are queueing messages for this bucket;
+ // If not 'null', we are queueing messages for this bucket
// otherwise, we are sending them through.
Queue<Message> messages = new ConcurrentLinkedQueue<>();
} catch (Exception e) {
logger.error("Exception in {}", this, e);
} finally {
- /*
- * cleanly leave state -- we want to make sure that messages
- * are processed in order, so the queue needs to remain until
- * it is empty
- */
- logger.info("{}: entering cleanup state", this);
- for ( ; ; ) {
- Message message = messages.poll();
- if (message == null) {
- // no messages left, but this could change
- synchronized (Bucket.this) {
- message = messages.poll();
- if (message == null) {
- // no messages left
- if (state == this) {
- if (owner == Server.getThisServer()) {
- // we can now exit the state
- state = null;
- stateChanged();
- } else {
- /*
- * We need a grace period before we can
- * remove the 'state' value (this can happen
- * if we receive and process the bulk data
- * before receiving official confirmation
- * that we are owner of the bucket.
- */
- messages = null;
- }
+ run_cleanup();
+ }
+ }
+
+ private void run_cleanup() {
+ /*
+ * cleanly leave state -- we want to make sure that messages
+ * are processed in order, so the queue needs to remain until
+ * it is empty
+ */
+ logger.info("{}: entering cleanup state", this);
+ for ( ; ; ) {
+ Message message = messages.poll();
+ if (message == null) {
+ // no messages left, but this could change
+ synchronized (Bucket.this) {
+ message = messages.poll();
+ if (message == null) {
+ // no messages left
+ if (state == this) {
+ if (owner == Server.getThisServer()) {
+ // we can now exit the state
+ state = null;
+ stateChanged();
+ } else {
+ /*
+ * We need a grace period before we can
+ * remove the 'state' value (this can happen
+ * if we receive and process the bulk data
+ * before receiving official confirmation
+ * that we are owner of the bucket.
+ */
+ messages = null;
}
- break;
}
+ break;
}
}
- // this doesn't work -- it ends up right back in the queue
- // if 'messages' is defined
- message.process();
}
- if (messages == null) {
- // this indicates we need to enter a grace period before cleanup,
- try {
- logger.info("{}: entering grace period before terminating",
- this);
- Thread.sleep(unconfirmedGracePeriod);
- } catch (InterruptedException e) {
- // we are exiting in any case
- Thread.currentThread().interrupt();
- } finally {
- synchronized (Bucket.this) {
- // Do we need to confirm that we really are the owner?
- // What does it mean if we are not?
- if (state == this) {
- state = null;
- stateChanged();
- }
+ // this doesn't work -- it ends up right back in the queue
+ // if 'messages' is defined
+ message.process();
+ }
+ if (messages == null) {
+ // this indicates we need to enter a grace period before cleanup,
+ try {
+ logger.info("{}: entering grace period before terminating",
+ this);
+ Thread.sleep(unconfirmedGracePeriod);
+ } catch (InterruptedException e) {
+ // we are exiting in any case
+ Thread.currentThread().interrupt();
+ } finally {
+ synchronized (Bucket.this) {
+ // Do we need to confirm that we really are the owner?
+ // What does it mean if we are not?
+ if (state == this) {
+ state = null;
+ stateChanged();
}
}
}
- logger.info("{}: exiting cleanup state", this);
}
+ logger.info("{}: exiting cleanup state", this);
}
/**
public String toString() {
return "Bucket.NewOwner(" + index + ")";
}
- }
- /**
- * Restore bucket data.
- *
- * @param obj deserialized bucket data
- */
- private void restoreBucketData(Object obj) {
- if (obj instanceof List) {
- for (Object entry : (List<?>)obj) {
- if (entry instanceof Restore) {
- // entry-specific 'restore' operation
- ((Restore)entry).restore(this.index);
- } else {
- logger.error("{}: Expected '{}' but got '{}'",
- this, Restore.class.getName(),
- entry.getClass().getName());
+ /**
+ * Restore bucket data.
+ *
+ * @param obj deserialized bucket data
+ */
+ private void restoreBucketData(Object obj) {
+ if (obj instanceof List) {
+ for (Object entry : (List<?>)obj) {
+ if (entry instanceof Restore) {
+ // entry-specific 'restore' operation
+ ((Restore)entry).restore(Bucket.this.index);
+ } else {
+ logger.error("{}: Expected '{}' but got '{}'",
+ this, Restore.class.getName(),
+ entry.getClass().getName());
+ }
}
+ } else {
+ logger.error("{}: expected 'List' but got '{}'",
+ this, obj.getClass().getName());
}
- } else {
- logger.error("{}: expected 'List' but got '{}'",
- this, obj.getClass().getName());
}
}
+
/* ============================================================ */
/**
@Override
public WebTarget webTarget(WebTarget webTarget) {
return webTarget
- .queryParam("bucket", index)
- .queryParam("dest", newOwner.getUuid())
- .queryParam("ttl", timeToLive);
+ .queryParam(QP_BUCKET, index)
+ .queryParam(QP_DEST, newOwner.getUuid())
+ .queryParam(QP_TTL, timeToLive);
}
@Override
import static org.onap.policy.drools.serverpool.ServerPoolProperties.getProperty;
import com.google.gson.Gson;
-import com.google.gson.JsonObject;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import static org.onap.policy.drools.serverpool.ServerPoolProperties.DEFAULT_BUCKET_TIME_TO_LIVE;
import static org.onap.policy.drools.serverpool.ServerPoolProperties.getProperty;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
-import java.util.Enumeration;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import lombok.AllArgsConstructor;
-import org.drools.core.definitions.InternalKnowledgePackage;
-import org.drools.core.impl.KnowledgeBaseImpl;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.FactHandle;
import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
import org.onap.policy.drools.system.PolicyEngine;
import org.onap.policy.drools.system.PolicyEngineConstants;
import org.onap.policy.drools.utils.Pair;
-import org.onap.policy.drools.utils.PropertyUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// used for JSON <-> String conversion
private static StandardCoder coder = new StandardCoder();
- private static final String configFile =
+ private static final String CONFIG_FILE =
"config/feature-server-pool.properties";
/*
private static long droolsTimeoutMillis;
private static String timeToLiveSecond;
+ // HTTP query parameters
+ private static final String QP_KEYWORD = "keyword";
+ private static final String QP_SESSION = "session";
+ private static final String QP_BUCKET = "bucket";
+ private static final String QP_TTL = "ttl";
+ private static final String QP_CONTROLLER = "controller";
+ private static final String QP_PROTOCOL = "protocol";
+ private static final String QP_TOPIC = "topic";
+
/******************************/
/* 'OrderedService' interface */
/******************************/
@Override
public boolean afterStart(PolicyEngine engine) {
logger.info("Starting FeatureServerPool");
- Server.startup(configFile);
+ Server.startup(CONFIG_FILE);
TargetLock.startup();
droolsTimeoutMillis =
getProperty(BUCKET_DROOLS_TIMEOUT, DEFAULT_BUCKET_DROOLS_TIMEOUT);
+ session.getName();
return webTarget
- .queryParam("keyword", keyword)
- .queryParam("session", encodedSessionName)
- .queryParam("bucket", bucketNumber)
- .queryParam("ttl", timeToLiveSecond);
+ .queryParam(QP_KEYWORD, keyword)
+ .queryParam(QP_SESSION, encodedSessionName)
+ .queryParam(QP_BUCKET, bucketNumber)
+ .queryParam(QP_TTL, timeToLiveSecond);
}
@Override
path[path.length - 1] = fieldName;
}
keyword = sco.getString(path);
+ if (keyword != null) {
+ if (conversionFunctionName == null) {
+ // We found a keyword -- we don't need to try other paths,
+ // so we should break out of the loop
+ break;
+ }
+
+ // we have post-processing to do
+ keyword = Keyword.convertKeyword(keyword, conversionFunctionName);
+ if (keyword != null) {
+ // conversion was successful
+ break;
+ }
+ }
}
if (keyword == null) {
}
}
}
- } else if ((ttl -= 1) > 0) {
- /*
- * This host is not the intended destination -- this could happen
- * if it was sent from another site. Forward the message in the
- * same thread.
- */
- forwardInsertDroolsMessage(bucket, keyword, sessionName, ttl, data);
+ } else {
+ ttl -= 1;
+ if (ttl > 0) {
+ /*
+ * This host is not the intended destination -- this could happen
+ * if it was sent from another site. Forward the message in the
+ * same thread.
+ */
+ forwardInsertDroolsMessage(bucket, keyword, sessionName, ttl, data);
+ }
}
}
Entity.entity(new String(data, StandardCharsets.UTF_8),
MediaType.APPLICATION_OCTET_STREAM_TYPE);
webTarget
- .queryParam("keyword", keyword)
- .queryParam("session", sessionName)
- .queryParam("bucket", bucket)
- .queryParam("ttl", ttl)
+ .queryParam(QP_KEYWORD, keyword)
+ .queryParam(QP_SESSION, sessionName)
+ .queryParam(QP_BUCKET, bucket)
+ .queryParam(QP_TTL, ttl)
.request().post(entity);
}
}
@Override
public WebTarget webTarget(WebTarget webTarget) {
return webTarget
- .queryParam("bucket", bucketNumber)
- .queryParam("keyword", keyword)
- .queryParam("controller", controller.getName())
- .queryParam("protocol", protocol.toString())
- .queryParam("topic", topic);
+ .queryParam(QP_BUCKET, bucketNumber)
+ .queryParam(QP_KEYWORD, keyword)
+ .queryParam(QP_CONTROLLER, controller.getName())
+ .queryParam(QP_PROTOCOL, protocol.toString())
+ .queryParam(QP_TOPIC, topic);
}
@Override
public void response(Response response) {
- // TODO: eventually, we will want to do something different
- // based upon success/failure
+ // log a message indicating success/failure
+ int status = response.getStatus();
+ if (status >= 200 && status <= 299) {
+ logger.info("/bucket/topic response code = {}", status);
+ } else {
+ logger.error("/bucket/topic response code = {}", status);
+ }
}
});
}
logger.info("{}: about to fetch data for session {}",
this, session.getFullName());
- kieSession.insert(new DroolsRunnable() {
- @Override
- public void run() {
- List<Object> droolsObjects = new ArrayList<>();
- for (FactHandle fh : kieSession.getFactHandles()) {
- Object obj = kieSession.getObject(fh);
- String keyword = Keyword.lookupKeyword(obj);
- if (keyword != null
- && Bucket.bucketNumber(keyword) == bucketNumber) {
- // bucket matches -- include this object
- droolsObjects.add(obj);
- /*
- * delete this factHandle from Drools memory
- * this classes are used in bucket migration,
- * so the delete is intentional.
- */
- kieSession.delete(fh);
- }
+ DroolsRunnable backupAndRemove = () -> {
+ List<Object> droolsObjects = new ArrayList<>();
+ for (FactHandle fh : kieSession.getFactHandles()) {
+ Object obj = kieSession.getObject(fh);
+ String keyword = Keyword.lookupKeyword(obj);
+ if (keyword != null
+ && Bucket.bucketNumber(keyword) == bucketNumber) {
+ // bucket matches -- include this object
+ droolsObjects.add(obj);
+ /*
+ * delete this factHandle from Drools memory
+ * this classes are used in bucket migration,
+ * so the delete is intentional.
+ */
+ kieSession.delete(fh);
}
-
- // send notification that object list is complete
- droolsObjectsWrapper.complete(droolsObjects);
}
- });
+
+ // send notification that object list is complete
+ droolsObjectsWrapper.complete(droolsObjects);
+ };
+ kieSession.insert(backupAndRemove);
// add pending operation to the list
pendingData.add(new Pair<>(droolsObjectsWrapper, session));
}
} catch (InterruptedException e) {
logger.error("Exception in {}", this, e);
+ Thread.currentThread().interrupt();
}
}
}
final KieSession kieSession = session.getKieSession();
// run the following within the Drools session thread
- kieSession.insert(new DroolsRunnable() {
- @Override
- public void run() {
- try {
- /*
- * Insert all of the objects -- note that this is running
- * in the session thread, so no other rules can fire
- * until all of the objects are inserted.
- */
- for (Object obj : droolsObjects) {
- kieSession.insert(obj);
- }
- } finally {
- // send notification that the inserts have completed
- sessionLatch.countDown();
+ DroolsRunnable doRestore = () -> {
+ try {
+ /*
+ * Insert all of the objects -- note that this is running
+ * in the session thread, so no other rules can fire
+ * until all of the objects are inserted.
+ */
+ for (Object droolsObj : droolsObjects) {
+ kieSession.insert(droolsObj);
}
+ } finally {
+ // send notification that the inserts have completed
+ sessionLatch.countDown();
}
- });
+ };
+ kieSession.insert(doRestore);
return sessionLatch;
} else {
logger.error("{}: Invalid session data for session={}, type={}",
// this table can be used to map an object class into the method
// to invoke to do the lookup
- private static ConcurrentHashMap<Class, Lookup> classToLookup =
+ private static ConcurrentHashMap<Class<?>, Lookup> classToLookup =
new ConcurrentHashMap<>();
// this is a pre-defined 'Lookup' instance that always returns 'null'
- private static Lookup nullLookup = new Lookup() {
- @Override
- public String getKeyword(Object obj) {
- return null;
- }
- };
+ private static Lookup nullLookup = (Object obj) -> (String) null;
/**
* This method takes the object's class, looks it up in the 'classToLookup'
// try to locate a matching entry using 'inheritance' rules
Class<?> thisClass = obj.getClass();
Class<?> matchingClass = null;
- for (Map.Entry<Class, Lookup> entry : classToLookup.entrySet()) {
+ for (Map.Entry<Class<?>, Lookup> entry : classToLookup.entrySet()) {
if (entry.getKey().isAssignableFrom(thisClass)
&& (matchingClass == null
|| matchingClass.isAssignableFrom(entry.getKey()))) {
}
}
- return lookupClassByName(classNameToSequence, clazz);
+ Class<?> keyClass = buildReflectiveLookup_findKeyClass(clazz);
+
+ if (keyClass == null) {
+ // no matching class name found
+ return null;
+ }
+
+ return buildReflectiveLookup_build(clazz, keyClass);
}
/**
* interfaces. If no match is found, repeat with the superclass,
* and all the way up the superclass chain.
*/
- private static Lookup lookupClassByName(Map<String, String> classNameToSequence,
- Class<?> clazz) {
+ private static Class<?> buildReflectiveLookup_findKeyClass(Class<?> clazz) {
Class<?> keyClass = null;
for (Class<?> cl = clazz ; cl != null ; cl = cl.getSuperclass()) {
if (classNameToSequence.containsKey(cl.getName())) {
break;
}
}
+ return keyClass;
+ }
- if (keyClass == null) {
- // no matching class name found
- return null;
- }
+ private static Lookup buildReflectiveLookup_build(Class<?> clazz, Class<?> keyClass) {
// we found a matching key in the table -- now, process the values
Class<?> currentClass = keyClass;
static final int UUID_LENGTH = 36;
static {
- conversionFunction.put("uuid", new Function<String, String>() {
- @Override
- public String apply(String value) {
- // truncate strings to 36 characters
- return value != null && value.length() > UUID_LENGTH
- ? value.substring(0, UUID_LENGTH) : value;
- }
+ conversionFunction.put("uuid", value -> {
+ // truncate strings to 36 characters
+ return value != null && value.length() > UUID_LENGTH
+ ? value.substring(0, UUID_LENGTH) : value;
});
}
// how may cycles of stability before declaring a winner
private static int stableVotingCycles;
+ /**
+ * Hide implicit public constructor.
+ */
+ private Leader() {
+ // everything here is static -- no instances of this class are created
+ }
+
/**
* Invoked at startup, or after some events -- immediately start a new vote.
*/
// decode base64 data
final byte[] packet = Base64.getDecoder().decode(data);
- MainLoop.queueWork(new Runnable() {
- /**
- * This method is running within the 'MainLoop' thread.
- */
- @Override
- public void run() {
- // create the 'VoteCycle' state machine, if needed
- if (voteCycle == null) {
- voteCycle = new VoteCycle();
- MainLoop.addBackgroundWork(voteCycle);
- }
- try {
- // pass data to 'VoteCycle' state machine
- voteCycle.packetReceived(packet);
- } catch (IOException e) {
- logger.error("Exception in 'Leader.voteData", e);
- }
+ MainLoop.queueWork(() -> {
+ // This runs within the 'MainLoop' thread --
+ // create the 'VoteCycle' state machine, if needed
+ if (voteCycle == null) {
+ voteCycle = new VoteCycle();
+ MainLoop.addBackgroundWork(voteCycle);
+ }
+ try {
+ // pass data to 'VoteCycle' state machine
+ voteCycle.packetReceived(packet);
+ } catch (IOException e) {
+ logger.error("Exception in 'Leader.voteData", e);
}
});
}
@Override
public void run() {
switch (state) {
- case STARTUP: {
- // 5-second grace period -- wait for things to stablize before
- // starting the vote
- if ((cycleCount -= 1) <= 0) {
- logger.info("VoteCycle: {} seconds have passed",
- stableIdleCycles);
- //MainLoop.removeBackgroundWork(this);
- updateMyVote();
- sendOutUpdates();
- state = State.VOTING;
- cycleCount = stableVotingCycles;
- }
+ case STARTUP:
+ startupState();
break;
- }
- case VOTING: {
- // need to be in the VOTING state without any vote changes
- // for 5 seconds -- once this happens, the leader is chosen
- if (sendOutUpdates()) {
- // changes have occurred -- set the grace period to 5 seconds
- cycleCount = stableVotingCycles;
- } else if ((cycleCount -= 1) <= 0) {
- // 5 second grace period has passed -- the leader is one with
- // the most votes, which is the first entry in 'voteData'
- Server oldLeader = leader;
- leader = Server.getServer(voteData.first().uuid);
- if (leader != oldLeader) {
- // the leader has changed -- send out notifications
- for (Events listener : Events.getListeners()) {
- listener.newLeader(leader);
- }
- } else {
- // the election is over, and the leader has been confirmed
- for (Events listener : Events.getListeners()) {
- listener.leaderConfirmed(leader);
- }
- }
- if (leader == Server.getThisServer()) {
- // this is the lead server --
- // make sure the 'Discovery' threads are running
- Discovery.startDiscovery();
- } else {
- // this is not the lead server -- stop 'Discovery' threads
- Discovery.stopDiscovery();
- }
-
- // we are done with voting -- clean up, and report results
- MainLoop.removeBackgroundWork(this);
- voteCycle = null;
-
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- PrintStream out = new PrintStream(bos);
-
- out.println("Voting results:");
-
- // x(36) xxxxx x(36)
- // UUID Votes Voter
- String format = "%-36s %5s %-36s\n";
- out.format(format, "UUID", "Votes", "Voter(s)");
- out.format(format, "----", "-----", "--------");
-
- for (VoteData vote : voteData) {
- if (vote.voters.isEmpty()) {
- out.format(format, vote.uuid, 0, "");
- } else {
- boolean headerNeeded = true;
- for (VoterData voter : vote.voters) {
- if (headerNeeded) {
- out.format(format, vote.uuid,
- vote.voters.size(), voter.uuid);
- headerNeeded = false;
- } else {
- out.format(format, "", "", voter.uuid);
- }
- }
- }
- }
-
- logger.info(bos.toString());
- }
+ case VOTING:
+ votingState();
break;
- }
+
default:
logger.error("Unknown state: {}", state);
break;
}
}
+ private void startupState() {
+ // 5-second grace period -- wait for things to stablize before
+ // starting the vote
+ cycleCount -= 1;
+ if (cycleCount <= 0) {
+ logger.info("VoteCycle: {} seconds have passed",
+ stableIdleCycles);
+ updateMyVote();
+ sendOutUpdates();
+ state = State.VOTING;
+ cycleCount = stableVotingCycles;
+ }
+ }
+
+ private void votingState() {
+ // need to be in the VOTING state without any vote changes
+ // for 5 seconds -- once this happens, the leader is chosen
+ if (sendOutUpdates()) {
+ // changes have occurred -- set the grace period to 5 seconds
+ cycleCount = stableVotingCycles;
+ return;
+ }
+
+ cycleCount -= 1;
+ if (cycleCount > 0) {
+ return;
+ }
+
+ // 5 second grace period has passed -- the leader is one with
+ // the most votes, which is the first entry in 'voteData'
+ Server oldLeader = leader;
+ leader = Server.getServer(voteData.first().uuid);
+ if (leader != oldLeader) {
+ // the leader has changed -- send out notifications
+ for (Events listener : Events.getListeners()) {
+ listener.newLeader(leader);
+ }
+ } else {
+ // the election is over, and the leader has been confirmed
+ for (Events listener : Events.getListeners()) {
+ listener.leaderConfirmed(leader);
+ }
+ }
+ if (leader == Server.getThisServer()) {
+ // this is the lead server --
+ // make sure the 'Discovery' threads are running
+ Discovery.startDiscovery();
+ } else {
+ // this is not the lead server -- stop 'Discovery' threads
+ Discovery.stopDiscovery();
+ }
+
+ // we are done with voting -- clean up, and report results
+ MainLoop.removeBackgroundWork(this);
+ voteCycle = null;
+
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ PrintStream out = new PrintStream(bos);
+
+ out.println("Voting results:");
+
+ // x(36) xxxxx x(36)
+ // UUID Votes Voter
+ String format = "%-36s %5s %-36s\n";
+ out.format(format, "UUID", "Votes", "Voter(s)");
+ out.format(format, "----", "-----", "--------");
+
+ for (VoteData vote : voteData) {
+ if (vote.voters.isEmpty()) {
+ out.format(format, vote.uuid, 0, "");
+ continue;
+ }
+ boolean headerNeeded = true;
+ for (VoterData voter : vote.voters) {
+ if (headerNeeded) {
+ out.format(format, vote.uuid,
+ vote.voters.size(), voter.uuid);
+ headerNeeded = false;
+ } else {
+ out.format(format, "", "", voter.uuid);
+ }
+ }
+ }
+
+ logger.info(bos.toString());
+ }
+
/**
* Process an incoming /vote REST message.
*
private void processVote(UUID voter, UUID vote, long timestamp) {
// fetch old data for this voter
VoterData voterData = uuidToVoterData.computeIfAbsent(voter,
- (key) -> new VoterData(voter, timestamp));
+ key -> new VoterData(voter, timestamp));
if (timestamp >= voterData.timestamp) {
// this is a new vote for this voter -- update the timestamp
voterData.timestamp = timestamp;
VoteData newVoteData = null;
if (vote != null) {
- newVoteData = uuidToVoteData.computeIfAbsent(vote, (key) -> new VoteData(vote));
+ newVoteData = uuidToVoteData.computeIfAbsent(vote, key -> new VoteData(vote));
}
if (oldVoteData != newVoteData) {
/**
* Poll for and process incoming messages for up to 1 second.
*/
- static void handleIncomingWork() throws InterruptedException {
- long currentTime = System.currentTimeMillis();;
+ static void handleIncomingWork() {
+ long currentTime = System.currentTimeMillis();
long wakeUpTime = currentTime + cycleTime;
long timeDiff;
work.run();
} catch (InterruptedException e) {
logger.error("Interrupted in MainLoop");
- throw(e);
+ Thread.currentThread().interrupt();
+ return;
} catch (Exception e) {
logger.error("Exception in MainLoop incoming work", e);
}
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
-import org.onap.policy.drools.serverpool.Bucket;
-import org.onap.policy.drools.serverpool.FeatureServerPool;
-import org.onap.policy.drools.serverpool.Leader;
-import org.onap.policy.drools.serverpool.Server;
-import org.onap.policy.drools.serverpool.TargetLock;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.text.SimpleDateFormat;
-import java.util.Arrays;
import java.util.Base64;
import java.util.Collection;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import javax.servlet.ServletException;
-import javax.ws.rs.ProcessingException;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
-import org.eclipse.jetty.server.ServerConnector;
import org.glassfish.jersey.client.ClientProperties;
import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
import org.onap.policy.common.endpoints.http.client.HttpClient;
import org.onap.policy.common.endpoints.http.client.HttpClientFactoryInstance;
import org.onap.policy.common.endpoints.http.server.HttpServletServer;
import org.onap.policy.common.endpoints.http.server.HttpServletServerFactoryInstance;
-import org.onap.policy.drools.system.PolicyEngineConstants;
import org.onap.policy.drools.utils.PropertyUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// the current REST server
private static HttpServletServer restServer;
- // incoming packets from HTTP
- private static LinkedTransferQueue<byte[]> incomingPackets =
- new LinkedTransferQueue<>();
-
/*==================================================*/
/* Some properties extracted at initialization time */
/*==================================================*/
static final int SOCKET_ADDRESS_TAG = 1;
static final int SITE_SOCKET_ADDRESS_TAG = 2;
+ // 'pingHosts' error
+ static final String PINGHOSTS_ERROR = "Server.pingHosts error";
+
/*==============================*/
/* Comparable<Server> interface */
/*==============================*/
InetAddress address = InetAddress.getByName(ipAddressString);
InetSocketAddress socketAddress = new InetSocketAddress(address, port);
+ possibleError = "HTTP server initialization error";
restServer = HttpServletServerFactoryInstance.getServerFactory().build(
"SERVER-POOL", // name
useHttps, // https
}
// we may not know the port until after the server is started
+ possibleError = "HTTP server start error";
restServer.start();
+ possibleError = null;
// determine the address to use
if (DEFAULT_SERVER_IP_ADDRESS.contentEquals(address.getHostAddress())) {
// start background thread
MainLoop.startThread();
- MainLoop.queueWork(new Runnable() {
- @Override
- public void run() {
- // run this in the 'MainLoop' thread
- Leader.startup();
- Bucket.startup();
- }
+ MainLoop.queueWork(() -> {
+ // run this in the 'MainLoop' thread
+ Leader.startup();
+ Bucket.startup();
});
logger.info("Listening on port {}", port);
int tag;
while ((tag = is.readUnsignedByte()) != END_OF_PARAMETERS_TAG) {
switch (tag) {
- case SOCKET_ADDRESS_TAG: {
+ case SOCKET_ADDRESS_TAG:
socketAddress = readSocketAddress(is);
break;
- }
- case SITE_SOCKET_ADDRESS_TAG: {
+ case SITE_SOCKET_ADDRESS_TAG:
siteSocketAddress = readSocketAddress(is);
break;
- }
default:
// ignore tag
logger.error("Illegal tag: {}", tag);
* @param is the 'DataInputStream'
* @return the 'InetSocketAddress'
*/
- private static InetSocketAddress readSocketAddress(DataInputStream is)
- throws IOException, UnknownHostException {
+ private static InetSocketAddress readSocketAddress(DataInputStream is) throws IOException {
byte[] ipAddress = new byte[4];
is.read(ipAddress, 0, 4);
return;
}
- getThreadPool().execute(new Runnable() {
+ getThreadPool().execute(() -> {
/**
* This method is running within the 'MainLoop' thread.
*/
- @Override
- public void run() {
- try {
- WebTarget webTarget = target.path(path);
- if (responseCallback != null) {
- // give callback a chance to modify 'WebTarget'
- webTarget = responseCallback.webTarget(webTarget);
-
- // send the response to the callback
- Response response;
- if (entity == null) {
- response = webTarget.request().get();
- } else {
- response = webTarget.request().post(entity);
- }
- responseCallback.response(response);
+ try {
+ WebTarget webTarget = target.path(path);
+ if (responseCallback != null) {
+ // give callback a chance to modify 'WebTarget'
+ webTarget = responseCallback.webTarget(webTarget);
+
+ // send the response to the callback
+ Response response;
+ if (entity == null) {
+ response = webTarget.request().get();
} else {
- // just do the invoke, and ignore the response
- if (entity == null) {
- webTarget.request().get();
- } else {
- webTarget.request().post(entity);
- }
+ response = webTarget.request().post(entity);
}
- } catch (Exception e) {
- logger.error("Failed to send to {} ({}, {})",
- uuid, destSocketAddress, destName);
+ responseCallback.response(response);
+ } else {
+ // just do the invoke, and ignore the response
+ if (entity == null) {
+ webTarget.request().get();
+ } else {
+ webTarget.request().post(entity);
+ }
+ }
+ } catch (Exception e) {
+ logger.error("Failed to send to {} ({}, {})",
+ uuid, destSocketAddress, destName);
+ if (responseCallback != null) {
responseCallback.exceptionResponse(e);
- MainLoop.queueWork(new Runnable() {
- @Override
- public void run() {
- // the DNS cache may have been out-of-date when this server
- // was first contacted -- fix the problem, if needed
- checkServer();
- }
- });
}
+ MainLoop.queueWork(() -> {
+ // this runs in the 'MainLoop' thread
+
+ // the DNS cache may have been out-of-date when this server
+ // was first contacted -- fix the problem, if needed
+ checkServer();
+ });
}
});
}
* in 'notifyList' (may need to build or rebuild 'notifyList').
*/
static void sendOutData() throws IOException {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(bos);
-
// include 'thisServer' in the data -- first, advance the count
- if ((thisServer.count += 1) == 0) {
+ thisServer.count += 1;
+ if (thisServer.count == 0) {
/*
- * counter wrapped (0 is a special case);
+ * counter wrapped (0 is a special case) --
* actually, we could probably leave this out, because it would take
* more than a century to wrap if the increment is 1 second
*/
thisServer.count = 1;
}
+
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(bos);
+
thisServer.lastUpdateTime = System.currentTimeMillis();
thisServer.writeServerData(dos);
}
} catch (NumberFormatException e) {
out.println(host + ": Invalid port value");
- logger.error("Server.pingHosts error", e);
+ logger.error(PINGHOSTS_ERROR, e);
error = true;
} catch (UnknownHostException e) {
out.println(host + ": Unknown host");
- logger.error("Server.pingHosts error", e);
+ logger.error(PINGHOSTS_ERROR, e);
error = true;
}
}
*/
static void pingHosts(final PrintStream out,
final Collection<InetSocketAddress> hosts) {
- FutureTask<Integer> ft = new FutureTask<>(new Callable<Integer>() {
- @Override
- public Integer call() {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(bos);
-
- // add information for this server only
- try {
- thisServer.writeServerData(dos);
-
- // create an 'Entity' that can be sent out to all hosts
- Entity<String> entity = Entity.entity(
- new String(Base64.getEncoder().encode(bos.toByteArray()),
- StandardCharsets.UTF_8),
- MediaType.APPLICATION_OCTET_STREAM_TYPE);
-
- // loop through hosts
- for (InetSocketAddress host : hosts) {
- HttpClient client = null;
-
- try {
- client = buildClient(host.toString(), host,
+ FutureTask<Integer> ft = new FutureTask<>(() -> {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(bos);
+
+ // add information for this server only
+ try {
+ thisServer.writeServerData(dos);
+
+ // create an 'Entity' that can be sent out to all hosts
+ Entity<String> entity = Entity.entity(
+ new String(Base64.getEncoder().encode(bos.toByteArray()),
+ StandardCharsets.UTF_8),
+ MediaType.APPLICATION_OCTET_STREAM_TYPE);
+
+ // loop through hosts
+ for (InetSocketAddress host : hosts) {
+ HttpClient httpClient = null;
+
+ try {
+ httpClient = buildClient(host.toString(), host,
socketAddressToName(host));
- getTarget(client).path("admin").request().post(entity);
- client.shutdown();
- client = null;
- } catch (KeyManagementException | NoSuchAlgorithmException e) {
- out.println(host + ": Unable to create client connection");
- logger.error("Server.pingHosts error", e);
- } catch (NoSuchFieldException | IllegalAccessException e) {
- out.println(host + ": Unable to get link to target");
- logger.error("Server.pingHosts error", e);
- } catch (Exception e) {
- out.println(host + ": " + e);
- logger.error("Server.pingHosts error", e);
- }
- if (client != null) {
- client.shutdown();
- }
+ getTarget(httpClient).path("admin").request().post(entity);
+ httpClient.shutdown();
+ httpClient = null;
+ } catch (KeyManagementException | NoSuchAlgorithmException e) {
+ out.println(host + ": Unable to create client connection");
+ logger.error(PINGHOSTS_ERROR, e);
+ } catch (NoSuchFieldException | IllegalAccessException e) {
+ out.println(host + ": Unable to get link to target");
+ logger.error(PINGHOSTS_ERROR, e);
+ } catch (Exception e) {
+ out.println(host + ": " + e);
+ logger.error(PINGHOSTS_ERROR, e);
+ }
+ if (httpClient != null) {
+ httpClient.shutdown();
}
- } catch (IOException e) {
- out.println("Unable to generate 'ping' data: " + e);
- logger.error("Server.pingHosts error", e);
}
- return 0;
+ } catch (IOException e) {
+ out.println("Unable to generate 'ping' data: " + e);
+ logger.error(PINGHOSTS_ERROR, e);
}
+ return 0;
});
MainLoop.queueWork(ft);
try {
ft.get(60, TimeUnit.SECONDS);
- } catch (InterruptedException | ExecutionException | TimeoutException e) {
+ } catch (InterruptedException e) {
+ logger.error("Server.pingHosts: interrupted waiting for queued work", e);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException | TimeoutException e) {
logger.error("Server.pingHosts: error waiting for queued work", e);
}
}
* @param out the 'PrintStream' to dump the table to
*/
public static void dumpHosts(final PrintStream out) {
- FutureTask<Integer> ft = new FutureTask<Integer>(new Callable<Integer>() {
- public Integer call() {
- dumpHostsInternal(out);
- return 0;
- }
+ FutureTask<Integer> ft = new FutureTask<>(() -> {
+ dumpHostsInternal(out);
+ return 0;
});
MainLoop.queueWork(ft);
try {
ft.get(60, TimeUnit.SECONDS);
- } catch (InterruptedException | ExecutionException | TimeoutException e) {
+ } catch (InterruptedException e) {
+ logger.error("Server.dumpHosts: interrupted waiting for queued work", e);
+ Thread.currentThread().interrupt();
+ } catch (ExecutionException | TimeoutException e) {
logger.error("Server.dumpHosts: error waiting for queued work", e);
}
}
} else if (localNotifyList.contains(server)) {
thisOne = "n";
}
- /*
- else if (newHosts.contains(server))
- {
- thisOne = "N";
- }
- */
if (siteData) {
String siteIp = "";
// save initial set of properties
private static Properties properties = new Properties();
+ /**
+ * Hide implicit public constructor.
+ */
+ private ServerPoolProperties() {
+ // everything here is static -- no instances of this class are created
+ }
+
/**
* Store the application properties values.
*
private static ReferenceQueue<TargetLock> abandoned = new ReferenceQueue<>();
// some status codes
- static final int ACCEPTED = 202; //Response.Status.ACCEPTED.getStatusCode();
- static final int NO_CONTENT = 204; //Response.Status.NO_CONTENT.getStatusCode();
+ static final int ACCEPTED = 202; //Response.Status.ACCEPTED.getStatusCode()
+ static final int NO_CONTENT = 204; //Response.Status.NO_CONTENT.getStatusCode()
static final int LOCKED = 423;
// Values extracted from properties
// this is used to notify the application when a lock is available,
// or if it is not available
- private LockCallback owner;
+ private volatile LockCallback owner;
// This is what is actually called by the infrastructure to do the owner
// notification. The owner may be running in a Drools session, in which case
// the actual notification should be done within that thread -- the 'context'
// object ensures that it happens this way.
- private LockCallback context;
+ private volatile LockCallback context;
+
+ // HTTP query parameters
+ private static final String QP_KEY = "key";
+ private static final String QP_OWNER = "owner";
+ private static final String QP_UUID = "uuid";
+ private static final String QP_WAIT = "wait";
+ private static final String QP_SERVER = "server";
+ private static final String QP_TTL = "ttl";
/**
* This method triggers registration of 'eventHandler', and also extracts
if (session != null) {
// deliver through a 'PolicySessionContext' class
Object lcontext = session.getAdjunct(PolicySessionContext.class);
- if (lcontext == null || !(lcontext instanceof LockCallback)) {
+ if (!(lcontext instanceof LockCallback)) {
context = new PolicySessionContext(session);
session.setAdjunct(PolicySessionContext.class, context);
} else {
@Override
public WebTarget webTarget(WebTarget webTarget) {
return webTarget
- .queryParam("key", key)
- .queryParam("owner", ownerKey)
- .queryParam("uuid", identity.uuid.toString())
- .queryParam("wait", waitForLock)
- .queryParam("ttl", timeToLive);
+ .queryParam(QP_KEY, key)
+ .queryParam(QP_OWNER, ownerKey)
+ .queryParam(QP_UUID, identity.uuid.toString())
+ .queryParam(QP_WAIT, waitForLock)
+ .queryParam(QP_TTL, timeToLive);
}
/**
* 423 (Locked) - lock in use, and 'waitForLock' is 'false'
*/
switch (response.getStatus()) {
- case NO_CONTENT: {
+ case NO_CONTENT:
// lock successful
setState(State.ACTIVE);
context.lockAvailable(TargetLock.this);
break;
- }
- case LOCKED: {
+ case LOCKED:
// failed -- lock in use, and 'waitForLock == false'
setState(State.FREE);
synchronized (localLocks) {
wr.clear();
context.lockUnavailable(TargetLock.this);
break;
- }
case ACCEPTED:
break;
default:
- logger.error("Unknown status: ", response.getStatus());
+ logger.error("Unknown status: {}", response.getStatus());
break;
}
}
*/
@Override
public void extend(int holdSec, LockCallback callback) {
+ // not implemented yet
}
/********************/
if (!Bucket.isKeyOnThisServer(key)) {
// this is the wrong server -- forward to the correct one
// (we can use this thread)
- if ((ttl -= 1) > 0) {
+ ttl -= 1;
+ if (ttl > 0) {
Server server = Bucket.bucketToServer(Bucket.bucketNumber(key));
if (server != null) {
WebTarget webTarget = server.getWebTarget("lock/lock");
server.getUuid(), key, ownerKey, uuid,
waitForLock, ttl);
return webTarget
- .queryParam("key", key)
- .queryParam("owner", ownerKey)
- .queryParam("uuid", uuid.toString())
- .queryParam("wait", waitForLock)
- .queryParam("ttl", String.valueOf(ttl))
+ .queryParam(QP_KEY, key)
+ .queryParam(QP_OWNER, ownerKey)
+ .queryParam(QP_UUID, uuid.toString())
+ .queryParam(QP_WAIT, waitForLock)
+ .queryParam(QP_TTL, String.valueOf(ttl))
.request().get();
}
}
if (!Bucket.isKeyOnThisServer(key)) {
// this is the wrong server -- forward to the correct one
// (we can use this thread)
- if ((ttl -= 1) > 0) {
+ ttl -= 1;
+ if (ttl > 0) {
Server server = Bucket.bucketToServer(Bucket.bucketNumber(key));
if (server != null) {
WebTarget webTarget = server.getWebTarget("lock/free");
+ "(key={},owner={},uuid={},ttl={})",
server.getUuid(), key, ownerKey, uuid, ttl);
return webTarget
- .queryParam("key", key)
- .queryParam("owner", ownerKey)
- .queryParam("uuid", uuid.toString())
- .queryParam("ttl", String.valueOf(ttl))
+ .queryParam(QP_KEY, key)
+ .queryParam(QP_OWNER, ownerKey)
+ .queryParam(QP_UUID, uuid.toString())
+ .queryParam(QP_TTL, String.valueOf(ttl))
.request().get();
}
}
if (!Bucket.isKeyOnThisServer(ownerKey)) {
// this is the wrong server -- forward to the correct one
// (we can use this thread)
- if ((ttl -= 1) > 0) {
+ ttl -= 1;
+ if (ttl > 0) {
Server server = Bucket.bucketToServer(Bucket.bucketNumber(key));
if (server != null) {
WebTarget webTarget = server.getWebTarget("lock/locked");
+ "(key={},owner={},uuid={},ttl={})",
server.getUuid(), key, ownerKey, uuid, ttl);
return webTarget
- .queryParam("key", key)
- .queryParam("owner", ownerKey)
- .queryParam("uuid", uuid.toString())
- .queryParam("ttl", String.valueOf(ttl))
+ .queryParam(QP_KEY, key)
+ .queryParam(QP_OWNER, ownerKey)
+ .queryParam(QP_UUID, uuid.toString())
+ .queryParam(QP_TTL, String.valueOf(ttl))
.request().get();
}
}
*/
@Override
public void shutdown() {
+ // nothing needs to be done
}
/**
@Override
public WebTarget webTarget(WebTarget webTarget) {
return webTarget
- .queryParam("key", key)
- .queryParam("owner", ownerKey)
- .queryParam("uuid", uuid.toString())
- .queryParam("ttl", timeToLive);
+ .queryParam(QP_KEY, key)
+ .queryParam(QP_OWNER, ownerKey)
+ .queryParam(QP_UUID, uuid.toString())
+ .queryParam(QP_TTL, timeToLive);
}
@Override
logger.info("Free response={} (code={})",
response, response.getStatus());
switch (response.getStatus()) {
- case NO_CONTENT: {
+ case NO_CONTENT:
// free successful -- don't need to do anything
break;
- }
- case LOCKED: {
+ case LOCKED:
// free failed
logger.error("TargetLock free failed, "
+ "key={}, owner={}, uuid={}",
key, ownerKey, uuid);
break;
- }
default:
- logger.error("Unknown status: ", response.getStatus());
+ logger.error("Unknown status: {}", response.getStatus());
break;
}
}
public void lockAvailable(final Lock lock) {
// Run 'owner.lockAvailable' within the Drools session
if (policySession != null) {
- policySession.getKieSession().insert(new DroolsRunnable() {
- @Override
- public void run() {
- ((TargetLock)lock).owner.lockAvailable(lock);
- }
- });
+ DroolsRunnable callback = () -> {
+ ((TargetLock)lock).owner.lockAvailable(lock);
+ };
+ policySession.getKieSession().insert(callback);
}
}
public void lockUnavailable(Lock lock) {
// Run 'owner.unlockAvailable' within the Drools session
if (policySession != null) {
- policySession.getKieSession().insert(new DroolsRunnable() {
- @Override
- public void run() {
- ((TargetLock)lock).owner.lockUnavailable(lock);
- }
- });
+ DroolsRunnable callback = () -> {
+ ((TargetLock)lock).owner.lockUnavailable(lock);
+ };
+ policySession.getKieSession().insert(callback);
}
}
*/
private static class LockEntry implements Serializable {
// string key identifying the lock
- String key;
+ private String key;
// string key identifying the owner
- String currentOwnerKey;
+ private String currentOwnerKey;
// UUID identifying the original 'TargetLock
- UUID currentOwnerUuid;
+ private UUID currentOwnerUuid;
// list of pending lock requests for this key
- Queue<Waiting> waitingList = new LinkedList<>();
+ private Queue<Waiting> waitingList = new LinkedList<>();
/**
* Constructor - initialize the 'LockEntry'.
@Override
public WebTarget webTarget(WebTarget webTarget) {
return webTarget
- .queryParam("key", key)
- .queryParam("owner", currentOwnerKey)
- .queryParam("uuid", currentOwnerUuid.toString())
- .queryParam("ttl", timeToLive);
+ .queryParam(QP_KEY, key)
+ .queryParam(QP_OWNER, currentOwnerKey)
+ .queryParam(QP_UUID, currentOwnerUuid.toString())
+ .queryParam(QP_TTL, timeToLive);
}
@Override
public void response(Response response) {
logger.info("Locked response={} (code={})",
response, response.getStatus());
- switch (response.getStatus()) {
- case NO_CONTENT: {
- // successful -- we are done
- break;
- }
-
- default: {
- // notification failed -- free this one
- globalLocks.unlock(key, currentOwnerUuid);
- break;
- }
+ if (response.getStatus() != NO_CONTENT) {
+ // notification failed -- free this one
+ globalLocks.unlock(key, currentOwnerUuid);
}
}
});
while (abandonedHandler != null) {
try {
Reference<? extends TargetLock> wr = abandoned.remove();
- TargetLock notify = null;
// At this point, we know that 'ref' is a
// 'WeakReference<TargetLock>' instance that has been abandoned,
*/
static byte[] dumpLocksData(UUID serverUuid, int ttl) throws IOException {
if (!Server.getThisServer().getUuid().equals(serverUuid)) {
- if ((ttl -= 1) > 0) {
+ ttl -= 1;
+ if (ttl > 0) {
Server server = Server.getServer(serverUuid);
if (server != null) {
WebTarget webTarget =
logger.info("Forwarding 'lock/dumpLocksData' to uuid {}",
serverUuid);
return webTarget
- .queryParam("server", serverUuid.toString())
- .queryParam("ttl", String.valueOf(ttl))
+ .queryParam(QP_SERVER, serverUuid.toString())
+ .queryParam(QP_TTL, String.valueOf(ttl))
.request().get(byte[].class);
}
}
@Override
public WebTarget webTarget(WebTarget webTarget) {
return webTarget
- .queryParam("server", server.getUuid().toString())
- .queryParam("ttl", timeToLive);
+ .queryParam(QP_SERVER, server.getUuid().toString())
+ .queryParam(QP_TTL, timeToLive);
}
@Override
// process the client-end data
for (ClientData clientData : hostData.clientDataList) {
- // 'true' if the bucket associated with this 'ClientData'
- // doesn't belong to the remote server, as far as we can tell
- boolean serverMismatch =
- Bucket.bucketToServer(clientData.bucketNumber) != server;
-
- // each 'ClientDataRecord' instance corresponds to an
- // active 'Identity' (TargetLock) instance
- for (ClientDataRecord cdr : clientData.clientDataRecords) {
- // update maximum 'key' and 'ownerKey' lengths
- updateKeyLength(cdr.identity.key);
- updateOwnerKeyLength(cdr.identity.ownerKey);
-
- // fetch UUID
- UUID uuid = cdr.identity.uuid;
-
- // fetch/generate 'MergeData' instance for this UUID
- MergedData md = mergedDataMap.get(uuid);
- if (md == null) {
- md = new MergedData(uuid);
- mergedDataMap.put(uuid, md);
- }
-
- // update 'MergedData.clientDataRecord'
- if (md.clientDataRecord == null) {
- md.clientDataRecord = cdr;
- } else {
- md.comment("Duplicate client entry for UUID");
- }
-
- if (serverMismatch) {
- // need to generate an additional error
- md.comment(server.toString()
- + "(client) does not own bucket "
- + clientData.bucketNumber);
- }
- }
+ populateLockData_clientData(clientData, server);
}
// process the server-end data
for (ServerData serverData : hostData.serverDataList) {
- // 'true' if the bucket associated with this 'ServerData'
- // doesn't belong to the remote server, as far as we can tell
- boolean serverMismatch =
- Bucket.bucketToServer(serverData.bucketNumber) != server;
-
- // each 'LockEntry' instance corresponds to the current holder
- // of a lock, and all requestors waiting for it to be freed
- for (LockEntry le : serverData.globalLocks.keyToEntry.values()) {
- // update maximum 'key' and 'ownerKey' lengths
- updateKeyLength(le.key);
- updateOwnerKeyLength(le.currentOwnerKey);
-
- // fetch uuid
- UUID uuid = le.currentOwnerUuid;
-
- // fetch/generate 'MergeData' instance for this UUID
- MergedData md = mergedDataMap.get(uuid);
- if (md == null) {
- md = new MergedData(uuid);
- mergedDataMap.put(uuid, md);
- }
+ populateLockData_serverData(serverData, server);
+ }
+ } else {
+ logger.error("TargetLock.DumpLocks.populateLockData: "
+ + "received data has class {}",
+ decodedData.getClass().getName());
+ }
+ }
- // update 'lockEntries' table entry
- if (lockEntries.get(le.key) != null) {
- md.comment("Duplicate server entry for key " + le.key);
- } else {
- lockEntries.put(le.key, le);
- }
+ private void populateLockData_clientData(ClientData clientData, Server server) {
+ // 'true' if the bucket associated with this 'ClientData'
+ // doesn't belong to the remote server, as far as we can tell
+ boolean serverMismatch =
+ Bucket.bucketToServer(clientData.bucketNumber) != server;
- // update 'MergedData.serverLockEntry'
- // (leave 'MergedData.serverWaiting' as 'null', because
- // this field is only used for waiting entries)
- if (md.serverLockEntry == null) {
- md.serverLockEntry = le;
- } else {
- md.comment("Duplicate server entry for UUID");
- }
+ // each 'ClientDataRecord' instance corresponds to an
+ // active 'Identity' (TargetLock) instance
+ for (ClientDataRecord cdr : clientData.clientDataRecords) {
+ // update maximum 'key' and 'ownerKey' lengths
+ updateKeyLength(cdr.identity.key);
+ updateOwnerKeyLength(cdr.identity.ownerKey);
- if (serverMismatch) {
- // need to generate an additional error
- md.comment(server.toString()
- + "(server) does not own bucket "
- + serverData.bucketNumber);
- }
+ // fetch UUID
+ UUID uuid = cdr.identity.uuid;
- // we need 'MergeData' entries for all waiting requests
- for (Waiting waiting : le.waitingList) {
- // update maximum 'ownerKey' length
- updateOwnerKeyLength(waiting.ownerKey);
+ // fetch/generate 'MergeData' instance for this UUID
+ MergedData md = mergedDataMap.get(uuid);
+ if (md == null) {
+ md = new MergedData(uuid);
+ mergedDataMap.put(uuid, md);
+ }
- // fetch uuid
- uuid = waiting.ownerUuid;
+ // update 'MergedData.clientDataRecord'
+ if (md.clientDataRecord == null) {
+ md.clientDataRecord = cdr;
+ } else {
+ md.comment("Duplicate client entry for UUID");
+ }
- // fetch/generate 'MergeData' instance for this UUID
- md = mergedDataMap.get(uuid);
- if (md == null) {
- md = new MergedData(uuid);
- mergedDataMap.put(uuid, md);
- }
+ if (serverMismatch) {
+ // need to generate an additional error
+ md.comment(server.toString()
+ + "(client) does not own bucket "
+ + clientData.bucketNumber);
+ }
+ }
+ }
- // update 'MergedData.serverLockEntry' and
- // 'MergedData.serverWaiting'
- if (md.serverLockEntry == null) {
- md.serverLockEntry = le;
- md.serverWaiting = waiting;
- } else {
- md.comment("Duplicate server entry for UUID");
- }
+ private void populateLockData_serverData(ServerData serverData, Server server) {
+ // 'true' if the bucket associated with this 'ServerData'
+ // doesn't belong to the remote server, as far as we can tell
+ boolean serverMismatch =
+ Bucket.bucketToServer(serverData.bucketNumber) != server;
- if (serverMismatch) {
- // need to generate an additional error
- md.comment(server.toString()
- + "(server) does not own bucket "
- + serverData.bucketNumber);
- }
- }
- }
+ // each 'LockEntry' instance corresponds to the current holder
+ // of a lock, and all requestors waiting for it to be freed
+ for (LockEntry le : serverData.globalLocks.keyToEntry.values()) {
+ // update maximum 'key' and 'ownerKey' lengths
+ updateKeyLength(le.key);
+ updateOwnerKeyLength(le.currentOwnerKey);
+
+ // fetch uuid
+ UUID uuid = le.currentOwnerUuid;
+
+ // fetch/generate 'MergeData' instance for this UUID
+ MergedData md = mergedDataMap.get(uuid);
+ if (md == null) {
+ md = new MergedData(uuid);
+ mergedDataMap.put(uuid, md);
+ }
+
+ // update 'lockEntries' table entry
+ if (lockEntries.get(le.key) != null) {
+ md.comment("Duplicate server entry for key " + le.key);
+ } else {
+ lockEntries.put(le.key, le);
+ }
+
+ // update 'MergedData.serverLockEntry'
+ // (leave 'MergedData.serverWaiting' as 'null', because
+ // this field is only used for waiting entries)
+ if (md.serverLockEntry == null) {
+ md.serverLockEntry = le;
+ } else {
+ md.comment("Duplicate server entry for UUID");
+ }
+
+ if (serverMismatch) {
+ // need to generate an additional error
+ md.comment(server.toString()
+ + "(server) does not own bucket "
+ + serverData.bucketNumber);
+ }
+
+ // we need 'MergeData' entries for all waiting requests
+ for (Waiting waiting : le.waitingList) {
+ populateLockData_serverData_waiting(
+ serverData, server, serverMismatch, le, waiting);
}
+ }
+ }
+
+ private void populateLockData_serverData_waiting(
+ ServerData serverData, Server server, boolean serverMismatch,
+ LockEntry le, Waiting waiting) {
+
+ // update maximum 'ownerKey' length
+ updateOwnerKeyLength(waiting.ownerKey);
+
+ // fetch uuid
+ UUID uuid = waiting.ownerUuid;
+
+ // fetch/generate 'MergeData' instance for this UUID
+ MergedData md = mergedDataMap.get(uuid);
+ if (md == null) {
+ md = new MergedData(uuid);
+ mergedDataMap.put(uuid, md);
+ }
+
+ // update 'MergedData.serverLockEntry' and
+ // 'MergedData.serverWaiting'
+ if (md.serverLockEntry == null) {
+ md.serverLockEntry = le;
+ md.serverWaiting = waiting;
} else {
- logger.error("TargetLock.DumpLocks.populateLockData: "
- + "received data has class "
- + decodedData.getClass().getName());
+ md.comment("Duplicate server entry for UUID");
+ }
+
+ if (serverMismatch) {
+ // need to generate an additional error
+ md.comment(server.toString()
+ + "(server) does not own bucket "
+ + serverData.bucketNumber);
}
}
out.printf(format, "---", "---------", "----", "-----", "--------");
}
+ dump_serverTable(out);
+ dump_clientOnlyEntries(out);
+ }
+
+ private void dump_serverTable(PrintStream out) {
// iterate over the server table
for (LockEntry le : lockEntries.values()) {
// fetch merged data
dumpMoreComments(out, md);
}
}
+ }
+ private void dump_clientOnlyEntries(PrintStream out) {
// client records that don't have matching server entries
for (MergedData md : clientOnlyEntries.values()) {
ClientDataRecord cdr = md.clientDataRecord;
*/
static class HostData implements Serializable {
// the UUID of the host sending the data
- UUID hostUuid;
+ private UUID hostUuid;
// all of the information derived from the 'LocalLocks' data
- List<ClientData> clientDataList;
+ private List<ClientData> clientDataList;
// all of the information derived from the 'GlobalLocks' data
- List<ServerData> serverDataList;
+ private List<ServerData> serverDataList;
/**
* Constructor - this goes through all of the lock tables,
*/
static class ClientData implements Serializable {
// number of the bucket
- int bucketNumber;
+ private int bucketNumber;
// all of the client locks within this bucket
- List<ClientDataRecord> clientDataRecords;
+ private List<ClientDataRecord> clientDataRecords;
/**
* Constructor - initially, there are no 'clientDataRecords'.
*/
static class ClientDataRecord implements Serializable {
// contains key, ownerKey, uuid
- Identity identity;
+ private Identity identity;
// state field of 'TargetLock'
// (may be 'null' if there is no 'TargetLock')
- State state;
+ private State state;
/**
* Constructor - initialize the fields.
*/
static class ServerData implements Serializable {
// number of the bucket
- int bucketNumber;
+ private int bucketNumber;
// server-side data associated with a single bucket
- GlobalLocks globalLocks;
+ private GlobalLocks globalLocks;
/**
* Constructor - initialize the fields.
*/
static class AuditData implements Serializable {
// sending UUID
- UUID hostUuid;
+ private UUID hostUuid;
// client records that currently exist, or records to be cleared
// (depending upon message) -- client/server is from the senders side
- List<Identity> clientData;
+ private List<Identity> clientData;
// server records that currently exist, or records to be cleared
// (depending upon message) -- client/server is from the senders side
- List<Identity> serverData;
+ private List<Identity> serverData;
/**
* Constructor - set 'hostUuid' to the current host, and start with
*/
AuditData() {
hostUuid = Server.getThisServer().getUuid();
- clientData = new ArrayList<Identity>();
- serverData = new ArrayList<Identity>();
+ clientData = new ArrayList<>();
+ serverData = new ArrayList<>();
}
/**
AuditData response = new AuditData();
// compare remote servers client data with our server data
+ generateResponse_clientEnd(response, includeWarnings);
+
+ // test server data
+ generateResponse_serverEnd(response, includeWarnings);
+
+ return response;
+ }
+
+ private void generateResponse_clientEnd(AuditData response, boolean includeWarnings) {
for (Identity identity : clientData) {
- // we are the server in this case
+ // remote end is the client, and we are the server
Bucket bucket = Bucket.getBucket(identity.key);
GlobalLocks globalLocks =
bucket.getAdjunctDontCreate(GlobalLocks.class);
// it was 'clientData' to the sender, but 'serverData' to us
response.serverData.add(identity);
}
+ }
- // test server data
+ private void generateResponse_serverEnd(AuditData response, boolean includeWarnings) {
for (Identity identity : serverData) {
- // we are the client in this case
+ // remote end is the server, and we are the client
Bucket bucket = Bucket.getBucket(identity.ownerKey);
LocalLocks localLocks =
bucket.getAdjunctDontCreate(LocalLocks.class);
}
response.clientData.add(identity);
}
-
- return response;
}
/**
* Run a single audit cycle.
*/
static void runAudit() {
- try {
- logger.info("Starting TargetLock audit");
- Audit audit = new Audit();
+ logger.info("Starting TargetLock audit");
+ Audit audit = new Audit();
- // populate 'auditMap' table
- audit.build();
+ // populate 'auditMap' table
+ audit.build();
- // send to all of the servers in 'auditMap' (may include this server)
- audit.send();
- } catch (InterruptedException e) {
- logger.error("TargetLock audit interrupted", e);
- Thread.currentThread().interrupt();
- }
+ // send to all of the servers in 'auditMap' (may include this server)
+ audit.send();
}
/**
// this needs to run in the 'MainLoop' thread, because it is dependent
// upon the list of servers, and our position in this list
- MainLoop.queueWork(new Runnable() {
- /**
- * {@inheritDoc}
- */
- @Override
- public void run() {
- // current list of servers
- Collection<Server> servers = Server.getServers();
+ MainLoop.queueWork(() -> {
+ // this runs in the 'MainLoop' thread
- // count of the number of servers
- int count = servers.size();
+ // current list of servers
+ Collection<Server> servers = Server.getServers();
- // will contain our position in this list
- int index = 0;
+ // count of the number of servers
+ int count = servers.size();
- // current server
- Server thisServer = Server.getThisServer();
+ // will contain our position in this list
+ int index = 0;
- for (Server server : servers) {
- if (server == thisServer) {
- break;
- }
- index += 1;
+ // current server
+ Server thisServer = Server.getThisServer();
+
+ for (Server server : servers) {
+ if (server == thisServer) {
+ break;
}
+ index += 1;
+ }
- // if index == count, we didn't find this server
- // (which shouldn't happen)
-
- if (index < count) {
- // The servers are ordered by UUID, and 'index' is this
- // server's position within the list. Suppose the period is
- // 60000 (60 seconds), and there are 5 servers -- the first one
- // will run the audit at 0 seconds after the minute, the next
- // at 12 seconds after the minute, then 24, 36, 48.
- long offset = (period * index) / count;
-
- // the earliest time we want the audit to run
- long time = System.currentTimeMillis() + gracePeriod;
- long startTime = time - (time % period) + offset;
- if (startTime <= time) {
- startTime += period;
+ // if index == count, we didn't find this server
+ // (which shouldn't happen)
+
+ if (index < count) {
+ // The servers are ordered by UUID, and 'index' is this
+ // server's position within the list. Suppose the period is
+ // 60000 (60 seconds), and there are 5 servers -- the first one
+ // will run the audit at 0 seconds after the minute, the next
+ // at 12 seconds after the minute, then 24, 36, 48.
+ long offset = (period * index) / count;
+
+ // the earliest time we want the audit to run
+ long time = System.currentTimeMillis() + gracePeriod;
+ long startTime = time - (time % period) + offset;
+ if (startTime <= time) {
+ startTime += period;
+ }
+ synchronized (Audit.class) {
+ if (timerTask != null) {
+ timerTask.cancel();
}
- synchronized (Audit.class) {
- if (timerTask != null) {
- timerTask.cancel();
+ timerTask = new TimerTask() {
+ @Override
+ public void run() {
+ runAudit();
}
- timerTask = new TimerTask() {
- @Override
- public void run() {
- runAudit();
- }
- };
+ };
- // now, schedule the timer
- Util.timer.scheduleAtFixedRate(
- timerTask, new Date(startTime), period);
- }
+ // now, schedule the timer
+ Util.timer.scheduleAtFixedRate(
+ timerTask, new Date(startTime), period);
}
}
});
*/
static byte[] incomingAudit(UUID serverUuid, int ttl, byte[] encodedData) {
if (!Server.getThisServer().getUuid().equals(serverUuid)) {
- if ((ttl -= 1) > 0) {
+ ttl -= 1;
+ if (ttl > 0) {
Server server = Server.getServer(serverUuid);
if (server != null) {
WebTarget webTarget = server.getWebTarget("lock/audit");
Entity.entity(new String(encodedData),
MediaType.APPLICATION_OCTET_STREAM_TYPE);
return webTarget
- .queryParam("server", serverUuid.toString())
- .queryParam("ttl", String.valueOf(ttl))
+ .queryParam(QP_SERVER, serverUuid.toString())
+ .queryParam(QP_TTL, String.valueOf(ttl))
.request().post(entity, byte[].class);
}
}
Bucket bucket = Bucket.getBucket(i);
// client data
- LocalLocks localLocks =
- bucket.getAdjunctDontCreate(LocalLocks.class);
- if (localLocks != null) {
- synchronized (localLocks) {
- // we have client data for this bucket
- for (Identity identity :
- localLocks.weakReferenceToIdentity.values()) {
- // find or create the 'AuditData' instance associated
- // with the server owning the 'key'
- AuditData auditData = getAuditData(identity.key);
- if (auditData != null) {
- auditData.clientData.add(identity);
- }
+ build_clientData(bucket);
+
+ // server data
+ build_serverData(bucket);
+ }
+ }
+
+ private void build_clientData(Bucket bucket) {
+ // client data
+ LocalLocks localLocks =
+ bucket.getAdjunctDontCreate(LocalLocks.class);
+ if (localLocks != null) {
+ synchronized (localLocks) {
+ // we have client data for this bucket
+ for (Identity identity :
+ localLocks.weakReferenceToIdentity.values()) {
+ // find or create the 'AuditData' instance associated
+ // with the server owning the 'key'
+ AuditData auditData = getAuditData(identity.key);
+ if (auditData != null) {
+ auditData.clientData.add(identity);
}
}
}
+ }
+ }
- // server data
- GlobalLocks globalLocks =
- bucket.getAdjunctDontCreate(GlobalLocks.class);
- if (globalLocks != null) {
- // we have server data for this bucket
- Map<String, LockEntry> keyToEntry = globalLocks.keyToEntry;
- synchronized (keyToEntry) {
- for (LockEntry le : keyToEntry.values()) {
+ private void build_serverData(Bucket bucket) {
+ // server data
+ GlobalLocks globalLocks =
+ bucket.getAdjunctDontCreate(GlobalLocks.class);
+ if (globalLocks != null) {
+ // we have server data for this bucket
+ Map<String, LockEntry> keyToEntry = globalLocks.keyToEntry;
+ synchronized (keyToEntry) {
+ for (LockEntry le : keyToEntry.values()) {
+ // find or create the 'AuditData' instance associated
+ // with the current 'ownerKey'
+ AuditData auditData = getAuditData(le.currentOwnerKey);
+ if (auditData != null) {
+ // create an 'Identity' entry, and add it to
+ // the list associated with the remote server
+ auditData.serverData.add(
+ new Identity(le.key, le.currentOwnerKey,
+ le.currentOwnerUuid));
+ }
+
+ for (Waiting waiting : le.waitingList) {
// find or create the 'AuditData' instance associated
- // with the current 'ownerKey'
- AuditData auditData = getAuditData(le.currentOwnerKey);
+ // with the waiting entry 'ownerKey'
+ auditData = getAuditData(waiting.ownerKey);
if (auditData != null) {
// create an 'Identity' entry, and add it to
// the list associated with the remote server
auditData.serverData.add(
- new Identity(le.key, le.currentOwnerKey,
- le.currentOwnerUuid));
- }
-
- for (Waiting waiting : le.waitingList) {
- // find or create the 'AuditData' instance associated
- // with the waiting entry 'ownerKey'
- auditData = getAuditData(waiting.ownerKey);
- if (auditData != null) {
- // create an 'Identity' entry, and add it to
- // the list associated with the remote server
- auditData.serverData.add(
- new Identity(le.key, waiting.ownerKey,
- waiting.ownerUuid));
- }
+ new Identity(le.key, waiting.ownerKey,
+ waiting.ownerUuid));
}
}
}
// map 'key -> bucket number', and then 'bucket number' -> 'server'
Server server = Bucket.bucketToServer(Bucket.bucketNumber(key));
if (server != null) {
- AuditData auditData = auditMap.get(server);
- if (auditData == null) {
- // doesn't exist yet -- create it
- auditData = new AuditData();
- auditMap.put(server, auditData);
- }
+ AuditData auditData =
+ auditMap.computeIfAbsent(server, sk -> new AuditData());
return auditData;
}
* Using the collected 'auditMap', send out the messages to all of the
* servers.
*/
- void send() throws InterruptedException {
+ void send() {
if (auditMap.isEmpty()) {
logger.info("TargetLock audit: no locks on this server");
} else {
}
for (final Server server : auditMap.keySet()) {
- // fetch audit data
- AuditData auditData = auditMap.get(server);
+ send_server(server);
+ }
+ }
- if (server == Server.getThisServer()) {
- // process this locally
- final AuditData respData = auditData.generateResponse(true);
- if (respData.clientData.isEmpty()
- && respData.serverData.isEmpty()) {
- // no mismatches
- logger.info("TargetLock.Audit.send: "
- + "no errors from self ({})", server);
- continue;
+ private void send_server(final Server server) {
+ // fetch audit data
+ AuditData auditData = auditMap.get(server);
+
+ if (server == Server.getThisServer()) {
+ // process this locally
+ final AuditData respData = auditData.generateResponse(true);
+ if (respData.clientData.isEmpty()
+ && respData.serverData.isEmpty()) {
+ // no mismatches
+ logger.info("TargetLock.Audit.send: "
+ + "no errors from self ({})", server);
+ return;
+ }
+
+ // do the rest in a separate thread
+ server.getThreadPool().execute(() -> {
+ // wait a few seconds, and see if we still know of these
+ // errors
+ if (AuditPostResponse.responseSupport(
+ respData, "self (" + server + ")",
+ "TargetLock.Audit.send")) {
+ // a return falue of 'true' either indicates the
+ // mismatches were resolved after a retry, or we
+ // received an interrupt, and need to abort
+ return;
}
- // do the rest in a separate thread
- server.getThreadPool().execute(new Runnable() {
- @Override
- public void run() {
- // wait a few seconds, and see if we still know of these
- // errors
- logger.info("TargetLock.Audit.send: "
- + "mismatches from self ({})", server);
- try {
- Thread.sleep(auditRetryDelay);
- } catch (InterruptedException e) {
- logger.error("TargetLock.Audit.send: Interrupted "
- + "handling audit response from self ({})",
- server);
- // just abort
- Thread.currentThread().interrupt();
- return;
- }
+ // any mismatches left in 'respData' are still issues
+ respData.processResponse(server);
+ });
+ return;
+ }
- // This will check against our own data -- any mismatches
- // mean that things have changed since we sent out the
- // first message. We will remove any mismatches from
- // 'respData', and see if there are any left.
- AuditData mismatches = respData.generateResponse(false);
-
- respData.serverData.removeAll(mismatches.clientData);
- respData.clientData.removeAll(mismatches.serverData);
-
- if (respData.clientData.isEmpty()
- && respData.serverData.isEmpty()) {
- // no mismatches --
- // there must have been transient issues on our side
- logger.info("TargetLock.Audit.send: "
- + "no mismatches from self "
- + "({}) after retry", server);
- return;
- }
+ // serialize
+ byte[] encodedData = auditData.encode();
+ if (encodedData == null) {
+ // error has already been displayed
+ return;
+ }
- // any mismatches left in 'respData' are still issues
- respData.processResponse(server);
- }
- });
- continue;
- }
+ // generate entity
+ Entity<String> entity =
+ Entity.entity(new String(encodedData),
+ MediaType.APPLICATION_OCTET_STREAM_TYPE);
- // serialize
- byte[] encodedData = auditData.encode();
- if (encodedData == null) {
- // error has already been displayed
- continue;
- }
+ server.post("lock/audit", entity, new AuditPostResponse(server));
+ }
+ }
- // generate entity
- Entity<String> entity =
- Entity.entity(new String(encodedData),
- MediaType.APPLICATION_OCTET_STREAM_TYPE);
+ static class AuditPostResponse implements Server.PostResponse {
+ private Server server;
- server.post("lock/audit", entity, new Server.PostResponse() {
- @Override
- public WebTarget webTarget(WebTarget webTarget) {
- // include the 'uuid' keyword
- return webTarget
- .queryParam("server", server.getUuid().toString())
- .queryParam("ttl", timeToLive);
- }
+ AuditPostResponse(Server server) {
+ this.server = server;
+ }
- @Override
- public void response(Response response) {
- // process the response here
- AuditData respData =
- AuditData.decode(response.readEntity(byte[].class));
- if (respData == null) {
- logger.error("TargetLock.Audit.send: "
- + "couldn't process response from {}",
- server);
- return;
- }
+ @Override
+ public WebTarget webTarget(WebTarget webTarget) {
+ // include the 'uuid' keyword
+ return webTarget
+ .queryParam(QP_SERVER, server.getUuid().toString())
+ .queryParam(QP_TTL, timeToLive);
+ }
- // if we reach this point, we got a response
- if (respData.clientData.isEmpty()
- && respData.serverData.isEmpty()) {
- // no mismatches
- logger.info("TargetLock.Audit.send: "
- + "no errors from {}", server);
- return;
- }
+ @Override
+ public void response(Response response) {
+ // process the response here
+ AuditData respData =
+ AuditData.decode(response.readEntity(byte[].class));
+ if (respData == null) {
+ logger.error("TargetLock.Audit.send: "
+ + "couldn't process response from {}",
+ server);
+ return;
+ }
- // wait a few seconds, and see if we still know of these
- // errors
- logger.info("TargetLock.Audit.send: mismatches from {}",
- server);
- try {
- Thread.sleep(auditRetryDelay);
- } catch (InterruptedException e) {
- logger.error("TargetLock.Audit.send: Interrupted "
- + "handling audit response from {}",
- server);
- // just abort
- Thread.currentThread().interrupt();
- return;
- }
+ // if we reach this point, we got a response
+ if (respData.clientData.isEmpty()
+ && respData.serverData.isEmpty()) {
+ // no mismatches
+ logger.info("TargetLock.Audit.send: "
+ + "no errors from {}", server);
+ return;
+ }
- // This will check against our own data -- any mismatches
- // mean that things have changed since we sent out the
- // first message. We will remove any mismatches from
- // 'respData', and see if there are any left.
- AuditData mismatches = respData.generateResponse(false);
-
- respData.serverData.removeAll(mismatches.clientData);
- respData.clientData.removeAll(mismatches.serverData);
-
- if (respData.clientData.isEmpty()
- && respData.serverData.isEmpty()) {
- // no mismatches --
- // there must have been transient issues on our side
- logger.info("TargetLock.Audit.send: no mismatches from "
- + "{} after retry", server);
- return;
- }
+ // wait a few seconds, and see if we still know of these
+ // errors
+ if (responseSupport(respData, server, "AuditPostResponse.response")) {
+ // a return falue of 'true' either indicates the mismatches
+ // were resolved after a retry, or we received an interrupt,
+ // and need to abort
+ return;
+ }
- // any mismatches left in 'respData' are still there --
- // hopefully, they are transient issues on the other side
- AuditData auditData = new AuditData();
- auditData.clientData = respData.serverData;
- auditData.serverData = respData.clientData;
-
- // serialize
- byte[] encodedData = auditData.encode();
- if (encodedData == null) {
- // error has already been displayed
- return;
- }
+ // any mismatches left in 'respData' are still there --
+ // hopefully, they are transient issues on the other side
+ AuditData auditData = new AuditData();
+ auditData.clientData = respData.serverData;
+ auditData.serverData = respData.clientData;
- // generate entity
- Entity<String> entity =
- Entity.entity(new String(encodedData),
- MediaType.APPLICATION_OCTET_STREAM_TYPE);
-
- // send new list to other end
- response = server
- .getWebTarget("lock/audit")
- .queryParam("server", server.getUuid().toString())
- .queryParam("ttl", timeToLive)
- .request().post(entity);
-
- respData = AuditData.decode(response.readEntity(byte[].class));
- if (respData == null) {
- logger.error("TargetLock.auditDataBuilder.send: "
- + "couldn't process response from {}",
- server);
- return;
- }
+ // serialize
+ byte[] encodedData = auditData.encode();
+ if (encodedData == null) {
+ // error has already been displayed
+ return;
+ }
- // if there are mismatches left, they are presumably real
- respData.processResponse(server);
- }
- });
+ // generate entity
+ Entity<String> entity =
+ Entity.entity(new String(encodedData),
+ MediaType.APPLICATION_OCTET_STREAM_TYPE);
+
+ // send new list to other end
+ response = server
+ .getWebTarget("lock/audit")
+ .queryParam(QP_SERVER, server.getUuid().toString())
+ .queryParam(QP_TTL, timeToLive)
+ .request().post(entity);
+
+ respData = AuditData.decode(response.readEntity(byte[].class));
+ if (respData == null) {
+ logger.error("TargetLock.auditDataBuilder.send: "
+ + "couldn't process response from {}",
+ server);
+ return;
+ }
+
+ // if there are mismatches left, they are presumably real
+ respData.processResponse(server);
+ }
+
+ // Handle mismatches indicated by an audit response -- a return value of
+ // 'true' indicates that there were no mismatches after a retry, or
+ // we received an interrupt. In either case, the caller returns.
+ private static boolean responseSupport(AuditData respData, Object serverString, String caller) {
+ logger.info("{}: mismatches from {}", caller, serverString);
+ try {
+ Thread.sleep(auditRetryDelay);
+ } catch (InterruptedException e) {
+ logger.error("{}: Interrupted handling audit response from {}",
+ caller, serverString);
+ // just abort
+ Thread.currentThread().interrupt();
+ return true;
+ }
+
+ // This will check against our own data -- any mismatches
+ // mean that things have changed since we sent out the
+ // first message. We will remove any mismatches from
+ // 'respData', and see if there are any left.
+ AuditData mismatches = respData.generateResponse(false);
+
+ respData.serverData.removeAll(mismatches.clientData);
+ respData.clientData.removeAll(mismatches.serverData);
+
+ if (respData.clientData.isEmpty()
+ && respData.serverData.isEmpty()) {
+ // no mismatches --
+ // there must have been transient issues on our side
+ logger.info("{}: no mismatches from {} after retry",
+ caller, serverString);
+ return true;
}
+
+ return false;
}
}
}
// create a shared 'Timer' instance
public static final Timer timer = new Timer("Server Pool Timer", true);
+ /**
+ * Hide implicit public constructor.
+ */
+ private Util() {
+ // everything here is static -- no instances of this class are created
+ }
+
/**
* Internally, UUID objects use two 'long' variables, and the default
* comparison is signed, which means the order for the first and 16th digit
* The following comparator uses the ordering '0123456789abcdef' for all
* digits.
*/
- public static final Comparator<UUID> uuidComparator =
- new Comparator<UUID>() {
- public int compare(UUID u1, UUID u2) {
- // compare most significant portion
- int rval = Long.compareUnsigned(u1.getMostSignificantBits(),
- u2.getMostSignificantBits());
- if (rval == 0) {
- // most significant portion matches --
- // compare least significant portion
- rval = Long.compareUnsigned(u1.getLeastSignificantBits(),
- u2.getLeastSignificantBits());
- }
- return rval;
- }
- };
+ public static final Comparator<UUID> uuidComparator = (UUID u1, UUID u2) -> {
+ // compare most significant portion
+ int rval = Long.compareUnsigned(u1.getMostSignificantBits(),
+ u2.getMostSignificantBits());
+ if (rval == 0) {
+ // most significant portion matches --
+ // compare least significant portion
+ rval = Long.compareUnsigned(u1.getLeastSignificantBits(),
+ u2.getLeastSignificantBits());
+ }
+ return rval;
+ };
/* ============================================================ */
try {
return IOUtils.toString(input, StandardCharsets.UTF_8);
} catch (IOException e) {
- // TODO Auto-generated catch block
logger.error("Util.inputStreamToString error", e);
return "";
}
package org.onap.policy.drools.serverpool.persistence;
-import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
public class Persistence implements PolicySessionFeatureApi, ServerPoolApi {
private static Logger logger = LoggerFactory.getLogger(Persistence.class);
+ // HTTP query parameters
+ private static final String QP_BUCKET = "bucket";
+ private static final String QP_SESSION = "session";
+ private static final String QP_COUNT = "count";
+ private static final String QP_DEST = "dest";
+
/***************************************/
/* 'PolicySessionFeatureApi' interface */
/***************************************/
* @param bucket the bucket containing the 'GlobalLocks' adjunct
* @param globalLocks the 'GlobalLocks' adjunct
*/
- private static void sendLockDataToBackups(Bucket bucket, GlobalLocks globalLocks) {
+ private static void sendLockDataToBackups(final Bucket bucket, final GlobalLocks globalLocks) {
final int bucketNumber = bucket.getIndex();
SenderBucketData sbd = bucket.getAdjunct(SenderBucketData.class);
int lockCount = 0;
for (final Server server : servers) {
if (server != null) {
// send out REST command
- server.getThreadPool().execute(new Runnable() {
- @Override
- public void run() {
- WebTarget webTarget =
- server.getWebTarget("persistence/lock");
- if (webTarget != null) {
- webTarget
- .queryParam("bucket", bucketNumber)
- .queryParam("count", count)
- .queryParam("dest", server.getUuid())
- .request().post(entity);
- }
+ server.getThreadPool().execute(() -> {
+ WebTarget webTarget =
+ server.getWebTarget("persistence/lock");
+ if (webTarget != null) {
+ webTarget
+ .queryParam(QP_BUCKET, bucketNumber)
+ .queryParam(QP_COUNT, count)
+ .queryParam(QP_DEST, server.getUuid())
+ .request().post(entity);
}
});
}
for (final Server server : servers) {
if (server != null) {
// send out REST command
- server.getThreadPool().execute(new Runnable() {
- @Override
- public void run() {
- WebTarget webTarget =
- server.getWebTarget("persistence/session");
- if (webTarget != null) {
- webTarget
- .queryParam("bucket",
- bucket.getIndex())
- .queryParam("session",
- encodedSessionName)
- .queryParam("count", count)
- .queryParam("dest", server.getUuid())
- .request().post(entity);
- }
+ server.getThreadPool().execute(() -> {
+ WebTarget webTarget =
+ server.getWebTarget("persistence/session");
+ if (webTarget != null) {
+ webTarget
+ .queryParam(QP_BUCKET,
+ bucket.getIndex())
+ .queryParam(QP_SESSION,
+ encodedSessionName)
+ .queryParam(QP_COUNT, count)
+ .queryParam(QP_DEST, server.getUuid())
+ .request().post(entity);
}
});
}
* @return the associated 'SenderSessionBucketData' instance
*/
synchronized SenderSessionBucketData getSessionData(PolicySession session) {
- // try to fetch the associated instance
- SenderSessionBucketData rval = sessionData.get(session);
- if (rval == null) {
- // it doesn't exist, so create one
- rval = new SenderSessionBucketData();
- sessionData.put(session, rval);
- }
- return rval;
+ return sessionData.computeIfAbsent(session, key -> new SenderSessionBucketData());
}
/**
* bucket owner fails.
*/
public static class ReceiverBucketData {
+ static final String RESTORE_BUCKET_ERROR =
+ "Persistence.ReceiverBucketData.restoreBucket: ";
+
// maps session name into encoded data
Map<String, ReceiverSessionBucketData> sessionData = new HashMap<>();
// one entry for each Drools session being restored --
// indicates when the restore is complete (restore runs within
// the Drools session thread)
+ List<CountDownLatch> sessionLatches = restoreBucket_droolsSessions();
+
+ // restore lock data
+ restoreBucket_locks(bucket);
+
+ // wait for all of the sessions to update
+ try {
+ for (CountDownLatch sessionLatch : sessionLatches) {
+ if (!sessionLatch.await(10000L, TimeUnit.MILLISECONDS)) {
+ logger.error("{}: timed out waiting for session latch",
+ this);
+ }
+ }
+ } catch (InterruptedException e) {
+ logger.error("Exception in {}", this, e);
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private List<CountDownLatch> restoreBucket_droolsSessions() {
List<CountDownLatch> sessionLatches = new LinkedList<>();
- for (String sessionName : sessionData.keySet()) {
+ for (Map.Entry<String, ReceiverSessionBucketData> entry : sessionData.entrySet()) {
+ String sessionName = entry.getKey();
+ ReceiverSessionBucketData rsbd = entry.getValue();
+
// [0]="<groupId>" [1]="<artifactId>", [2]="<sessionName>"
String[] nameSegments = sessionName.split(":");
PolicySession policySession = null;
}
if (policySession == null) {
- logger.error("Persistence.ReceiverBucketData.restoreBucket: "
+ logger.error(RESTORE_BUCKET_ERROR
+ "Can't find PolicySession{}", sessionName);
continue;
}
Object obj = null;
try {
// deserialization needs to use the correct 'ClassLoader'
- ReceiverSessionBucketData rsbd = sessionData.get(sessionName);
obj = Util.deserialize(Base64.getDecoder().decode(rsbd.encodedSerializedData),
policySession.getPolicyContainer().getClassLoader());
} catch (IOException | ClassNotFoundException | IllegalArgumentException e) {
- logger.error("Persistence.ReceiverBucketData.restoreBucket: "
+ logger.error(RESTORE_BUCKET_ERROR
+ "Failed to read data for session '{}'",
sessionName, e);
}
if (!(obj instanceof Map)) {
- logger.error("Persistence.ReceiverBucketData.restoreBucket: "
+ logger.error(RESTORE_BUCKET_ERROR
+ "Session '{}' data has class {}, expected 'Map'",
sessionName, obj.getClass().getName());
final KieSession kieSession = policySession.getKieSession();
// run the following within the Drools session thread
- kieSession.insert(new DroolsRunnable() {
- /**
- * {@inheritDoc}
- */
- @Override
- public void run() {
- try {
- // insert all of the Drools objects into the session
- for (Object obj : droolsObjects.keySet()) {
- kieSession.insert(obj);
- }
- } finally {
- // signal completion
- sessionLatch.countDown();
+ DroolsRunnable insertDroolsObjects = () -> {
+ try {
+ // insert all of the Drools objects into the session
+ for (Object droolsObj : droolsObjects.keySet()) {
+ kieSession.insert(droolsObj);
}
+ } finally {
+ // signal completion
+ sessionLatch.countDown();
}
- });
+ };
+ kieSession.insert(insertDroolsObjects);
// add this to the set of 'CountDownLatch's we are waiting for
sessionLatches.add(sessionLatch);
}
+ return sessionLatches;
+ }
- // restore lock data
+ private void restoreBucket_locks(Bucket bucket) {
if (lockData != null) {
Object obj = null;
try {
// send out updated date
sendLockDataToBackups(bucket, (GlobalLocks)obj);
} else {
- logger.error("Persistence.ReceiverBucketData.restoreBucket: "
+ logger.error(RESTORE_BUCKET_ERROR
+ "Expected 'GlobalLocks', got '{}'",
obj.getClass().getName());
}
} catch (IOException | ClassNotFoundException | IllegalArgumentException e) {
- logger.error("Persistence.ReceiverBucketData.restoreBucket: "
+ logger.error(RESTORE_BUCKET_ERROR
+ "Failed to read lock data", e);
// skip the lock data
}
}
-
- // wait for all of the sessions to update
- try {
- for (CountDownLatch sessionLatch : sessionLatches) {
- if (!sessionLatch.await(10000L, TimeUnit.MILLISECONDS)) {
- logger.error("{}: timed out waiting for session latch",
- this);
- }
- }
- } catch (InterruptedException e) {
- logger.error("Exception in {}", this, e);
- Thread.currentThread().interrupt();
- }
}
}
@POST
@Path("/persistence/session")
@Consumes(MediaType.APPLICATION_OCTET_STREAM)
- public void receiveSession(@QueryParam("bucket") int bucket,
- @QueryParam("session") String sessionName,
- @QueryParam("count") int count,
- @QueryParam("dest") UUID dest,
+ public void receiveSession(@QueryParam(QP_BUCKET) int bucket,
+ @QueryParam(QP_SESSION) String sessionName,
+ @QueryParam(QP_COUNT) int count,
+ @QueryParam(QP_DEST) UUID dest,
byte[] data) {
logger.debug("/persistence/session: (bucket={},session={},count={}) "
+ "got {} bytes of data",
Entity.entity(new String(data),
MediaType.APPLICATION_OCTET_STREAM_TYPE);
webTarget
- .queryParam("bucket", bucket)
- .queryParam("session", sessionName)
- .queryParam("count", count)
+ .queryParam(QP_BUCKET, bucket)
+ .queryParam(QP_SESSION, sessionName)
+ .queryParam(QP_COUNT, count)
.request().post(entity);
}
}
@POST
@Path("/persistence/lock")
@Consumes(MediaType.APPLICATION_OCTET_STREAM)
- public void receiveLockData(@QueryParam("bucket") int bucket,
- @QueryParam("count") int count,
- @QueryParam("dest") UUID dest,
+ public void receiveLockData(@QueryParam(QP_BUCKET) int bucket,
+ @QueryParam(QP_COUNT) int count,
+ @QueryParam(QP_DEST) UUID dest,
byte[] data) {
logger.debug("/persistence/lock: (bucket={},count={}) "
+ "got {} bytes of data", bucket, count, data.length);
Entity.entity(new String(data),
MediaType.APPLICATION_OCTET_STREAM_TYPE);
webTarget
- .queryParam("bucket", bucket)
- .queryParam("count", count)
+ .queryParam(QP_BUCKET, bucket)
+ .queryParam(QP_COUNT, count)
.request().post(entity);
}
}
Code Review / policy / drools-pdp.git / commitdiff