X-Git-Url: https://gerrit.onap.org/r/gitweb?p=dmaap%2Fdatarouter.git;a=blobdiff_plain;f=datarouter-node%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fdmaap%2Fdatarouter%2Fnode%2FDeliveryQueue.java;h=4d336b733aebf7f2943c16baa5f2ab79e4027751;hp=8cdafd6f9b4f01ebc0eb09e4ed36030344239c8f;hb=8280a5e1a6cfd155edfafe882439b8c04c6a4509;hpb=52c5c5ab9ba33755e26f06f41de608f825866a53
diff --git a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java
index 8cdafd6f..4d336b73 100644
--- a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java
+++ b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java
@@ -27,51 +27,60 @@ package org.onap.dmaap.datarouter.node;
import com.att.eelf.configuration.EELFLogger;
import com.att.eelf.configuration.EELFManager;
import java.io.File;
+import java.util.ArrayList;
import java.util.Arrays;
-import java.util.Hashtable;
-import java.util.Vector;
+import java.util.HashMap;
+import java.util.List;
import org.jetbrains.annotations.Nullable;
/**
* Mechanism for monitoring and controlling delivery of files to a destination.
*
- * The DeliveryQueue class maintains lists of DeliveryTasks for a single destination (a subscription or another data
- * router node) and assigns delivery threads to try to deliver them. It also maintains a delivery status that causes it
- * to back off on delivery attempts after a failure.
+ *
The DeliveryQueue class maintains lists of DeliveryTasks for a single
+ * destination (a subscription or another data router node) and assigns
+ * delivery threads to try to deliver them. It also maintains a delivery
+ * status that causes it to back off on delivery attempts after a failure.
*
- *
If the most recent delivery result was a failure, then no more attempts will be made for a period of time.
- * Initially, and on the first failure following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer()
- * (milliseconds). If, after this delay, additional failures occur, each failure will multiply the delay by
- * DeliveryQueueHelper.getFailureBackoff() up to a maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer().
- * Note that this behavior applies to the delivery queue as a whole and not to individual files in the queue. If
- * multiple files are being delivered and one fails, the delay will be started. If a second delivery fails while the
- * delay was active, it will not change the delay or change the duration of any subsequent delay. If, however, it
- * succeeds, it will cancel the delay.
+ *
If the most recent delivery result was a failure, then no more attempts
+ * will be made for a period of time. Initially, and on the first failure
+ * following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() (milliseconds).
+ * If, after this delay, additional failures occur, each failure will
+ * multiply the delay by DeliveryQueueHelper.getFailureBackoff() up to a
+ * maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer().
+ * Note that this behavior applies to the delivery queue as a whole and not
+ * to individual files in the queue. If multiple files are being
+ * delivered and one fails, the delay will be started. If a second
+ * delivery fails while the delay was active, it will not change the delay
+ * or change the duration of any subsequent delay.
+ * If, however, it succeeds, it will cancel the delay.
+ * The queue maintains 3 collections of files to deliver: A todoList of
+ * files that will be attempted, a working set of files that are being
+ * attempted, and a retry set of files that were attempted and failed.
+ * Whenever the todoList is empty and needs to be refilled, a scan of the
+ * spool directory is made and the file names sorted. Any files in the working set are ignored.
+ * If a DeliveryTask for the file is in the retry set, then that delivery
+ * task is placed on the todoList. Otherwise, a new DeliveryTask for the
+ * file is created and placed on the todoList.
+ * If, when a DeliveryTask is about to be removed from the todoList, its
+ * age exceeds DeliveryQueueHelper.getExpirationTimer(), then it is instead
+ * marked as expired.
*
- *
The queue maintains 3 collections of files to deliver: A to do list of files that will be attempted, a working
- * set of files that are being attempted, and a retry set of files that were attempted and failed. Whenever the to do
- * list is empty and needs to be refilled, a scan of the spool directory is made and the file names sorted. Any files
- * in the working set are ignored. If a DeliveryTask for the file is in the retry set, then that delivery task is placed
- * on the to do list. Otherwise, a new DeliveryTask for the file is created and placed on the to do list. If, when a
- * DeliveryTask is about to be removed from the to do list, its age exceeds DeliveryQueueHelper.getExpirationTimer(),
- * then it is instead marked as expired.
- *
- *
A delivery queue also maintains a skip flag. This flag is true if the failure timer is active or if no files are
- * found in a directory scan.
+ *
A delivery queue also maintains a skip flag. This flag is true if the
+ * failure timer is active or if no files are found in a directory scan.
*/
public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
-
private static EELFLogger logger = EELFManager.getInstance().getLogger(DeliveryQueue.class);
private DeliveryQueueHelper deliveryQueueHelper;
+
private DestInfo destinationInfo;
- private Hashtable working = new Hashtable<>();
- private Hashtable retry = new Hashtable<>();
+ private HashMap working = new HashMap<>();
+ private HashMap retry = new HashMap<>();
private int todoindex;
private boolean failed;
private long failduration;
private long resumetime;
private File dir;
- private Vector todo = new Vector<>();
+ private List todoList = new ArrayList<>();
/**
* Create a delivery queue for a given destination info.
@@ -94,8 +103,8 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
DeliveryTask dt = retry.get(pubid);
if (dt == null) {
- for (int i = todoindex; i < todo.size(); i++) {
- DeliveryTask xdt = todo.get(i);
+ for (int i = todoindex; i < todoList.size(); i++) {
+ DeliveryTask xdt = todoList.get(i);
if (xdt.getPublishId().equals(pubid)) {
dt = xdt;
break;
@@ -111,8 +120,8 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
if (dt.isCleaned()) {
return (0);
}
- StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(),
- dt.getLength(), "diskFull", dt.getAttempts());
+ StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(),
+ dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts());
dt.clean();
return (dt.getLength());
}
@@ -122,6 +131,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
*/
private synchronized void markSuccess(DeliveryTask task) {
working.remove(task.getPublishId());
+ logger.info(task.getPublishId() + " marked as success.");
task.clean();
failed = false;
failduration = 0;
@@ -131,6 +141,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
* Mark that a delivery task has expired.
*/
private synchronized void markExpired(DeliveryTask task) {
+ logger.info(task.getPublishId() + " marked as expired.");
task.clean();
}
@@ -139,6 +150,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
*/
private synchronized void markFailNoRetry(DeliveryTask task) {
working.remove(task.getPublishId());
+ logger.info(task.getPublishId() + " marked as failed permanently");
task.clean();
failed = false;
failduration = 0;
@@ -168,6 +180,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
*/
private synchronized void markRedirect(DeliveryTask task) {
working.remove(task.getPublishId());
+ logger.info(task.getPublishId() + " marked as redirected.");
retry.put(task.getPublishId(), task);
}
@@ -176,6 +189,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
*/
private synchronized void markFailWithRetry(DeliveryTask task) {
working.remove(task.getPublishId());
+ logger.info(task.getPublishId() + " marked as temporarily failed.");
retry.put(task.getPublishId(), task);
fdupdate();
}
@@ -206,19 +220,20 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
}
while (true) {
- if (todoindex >= todo.size()) {
+ if (todoindex >= todoList.size()) {
todoindex = 0;
- todo = new Vector<>();
+ todoList = new ArrayList<>();
String[] files = dir.list();
Arrays.sort(files);
scanForNextTask(files);
- retry = new Hashtable<>();
+ retry = new HashMap<>();
}
DeliveryTask dt = getDeliveryTask(mindate);
if (dt != null) {
return dt;
}
return null;
+
}
}
@@ -267,8 +282,8 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid);
if (destinationInfo.isPrivilegedSubscriber()) {
- task.setResumeTime(
- System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
+ task.setResumeTime(System.currentTimeMillis()
+ + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
markFailWithRetry(task);
} else {
markSuccess(task);
@@ -279,9 +294,8 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
if (deliveryQueueHelper.handleRedirection(destinationInfo, location, task.getFileId())) {
markRedirect(task);
} else {
- StatusLog
- .logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
- task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
+ StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(),
+ task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
markFailNoRetry(task);
}
} else if (status < 500 && status != 429) {
@@ -327,13 +341,15 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
/**
- * Deliver files until there's a failure or there are no more files to deliver.
+ * Deliver files until there's a failure or there are no more.
+ * files to deliver
*/
public void run() {
DeliveryTask task;
long endtime = System.currentTimeMillis() + deliveryQueueHelper.getFairTimeLimit();
int filestogo = deliveryQueueHelper.getFairFileLimit();
while ((task = getNext()) != null) {
+ logger.info("Processing file: " + task.getPublishId());
task.run();
if (--filestogo <= 0 || System.currentTimeMillis() > endtime) {
break;
@@ -342,7 +358,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
/**
- * Is there no work to do for this queue right now.
+ * Is there no work to do for this queue right now?.
*/
synchronized boolean isSkipSet() {
return (peekNext() == null);
@@ -385,14 +401,14 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
if (dt == null) {
dt = new DeliveryTask(this, pubId);
}
- todo.add(dt);
+ todoList.add(dt);
}
}
@Nullable
private DeliveryTask getDeliveryTask(long mindate) {
- if (todoindex < todo.size()) {
- DeliveryTask dt = todo.get(todoindex);
+ if (todoindex < todoList.size()) {
+ DeliveryTask dt = todoList.get(todoindex);
if (dt.isCleaned()) {
todoindex++;
}
@@ -433,4 +449,4 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
return fname2;
}
-}
+}
\ No newline at end of file