> pubQueue, String retryKey, int retryLimit,
+ final String invalidKey,
+ final ManagedDataStore dataStore, final Codec codec,
+ final Runnable disconnectCallback, final WorkerConfiguration workerConfiguration)
{
this.callback = Objects.requireNonNull(callback);
this.metrics = Objects.requireNonNull(metrics);
@@ -63,57 +106,247 @@ public WorkerQueueConsumerImpl(TaskCallback callback, RabbitMetricsReporter metr
this.publisherEventQueue = Objects.requireNonNull(pubQueue);
this.retryRoutingKey = Objects.requireNonNull(retryKey);
this.retryLimit = retryLimit;
+ this.invalidRoutingKey = Objects.requireNonNull(invalidKey);
+ this.dataStore = Objects.requireNonNull(dataStore);
+ this.codec = Objects.requireNonNull(codec);
+ this.disconnectCallback = Objects.requireNonNull(disconnectCallback);
+ this.offloadedPayloadsToDelete = Collections.synchronizedSortedMap(new TreeMap<>());
+ this.workerConfiguration = Objects.requireNonNull(workerConfiguration);
}
/**
* {@inheritDoc}
- *
+ *
* If an incoming message is marked as redelivered, hand it off to another method to deal with retry/rejection. Otherwise, hand it off
- * to worker-core, and potentially repbulish or reject it depending upon exceptions thrown.
+ * to worker-core, and potentially republish or reject it depending upon exceptions thrown.
*/
@Override
public void processDelivery(Delivery delivery)
{
- final int retries = delivery.getHeaders().containsKey(RabbitHeaders.RABBIT_HEADER_CAF_DELIVERY_COUNT) ?
- Integer.parseInt(String.valueOf(delivery.getHeaders()
- .getOrDefault(RabbitHeaders.RABBIT_HEADER_CAF_DELIVERY_COUNT, "0"))) :
- Integer.parseInt(String.valueOf(delivery.getHeaders()
- .getOrDefault(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_RETRY, "0")));
-
+ final long inboundMessageId = delivery.getEnvelope().getDeliveryTag();
+ final String routingKey = delivery.getEnvelope().getRoutingKey();
+ final Map deliveryHeaders = delivery.getHeaders();
+ final boolean isRedelivered = delivery.getEnvelope().isRedeliver();
+ final int retries = deliveryHeaders.containsKey(RabbitHeaders.RABBIT_HEADER_CAF_DELIVERY_COUNT)
+ ? Integer.parseInt(String.valueOf(deliveryHeaders.getOrDefault(RabbitHeaders.RABBIT_HEADER_CAF_DELIVERY_COUNT, "0")))
+ : Integer.parseInt(String.valueOf(deliveryHeaders.getOrDefault(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_RETRY, "0")));
+ final Optional taskMessageStorageRefOpt
+ = Optional.ofNullable(deliveryHeaders.get(RABBIT_HEADER_CAF_PAYLOAD_OFFLOADING_STORAGE_REF)).map(Object::toString);
metrics.incrementReceived();
- final boolean isPoison;
- if (delivery.getEnvelope().isRedeliver()) {
- if (!delivery.getHeaders().containsKey(RabbitHeaders.RABBIT_HEADER_CAF_DELIVERY_COUNT)) {
- //RABBIT_HEADER_CAF_DELIVERY_COUNT is not available, message was delivered from CLASSIC queue
+
+ final byte[] deliveryMessageData = delivery.getMessageData();
+ final TaskMessage taskMessage;
+ try {
+ try {
+ taskMessage = codec.deserialise(deliveryMessageData, TaskMessage.class, DecodeMethod.LENIENT);
+ } catch (final CodecException e) {
+ handleInvalidDelivery(inboundMessageId, Optional.empty(), deliveryMessageData, deliveryHeaders,
+ "Cannot deserialize delivery messageData to TaskMessage");
+ return;
+ }
+
+ try {
+ handleTaskDataInjection(taskMessage, inboundMessageId, taskMessageStorageRefOpt);
+ } catch (final InvalidDeliveryException ex) {
+ handleInvalidDelivery(inboundMessageId, Optional.of(taskMessage), deliveryMessageData, deliveryHeaders,
+ ex.getMessage());
+ return;
+ }
+
+ final PoisonMessageStatus poisonMessageStatus = getPoisonMessageStatus(
+ isRedelivered, deliveryHeaders, retries);
+
+ if (poisonMessageStatus == PoisonMessageStatus.CLASSIC_POSSIBLY_POISON) {
+ republishClassicRedelivery(
+ delivery.getEnvelope().getRoutingKey(),
+ inboundMessageId,
+ deliveryMessageData,
+ taskMessage.getTaskData(),
+ deliveryHeaders,
+ retries,
+ taskMessage.getTracking(),
+ taskMessageStorageRefOpt
+ );
+ return;
+ }
+
+ processDelivery(
+ inboundMessageId,
+ routingKey,
+ deliveryHeaders,
+ taskMessage,
+ deliveryMessageData,
+ poisonMessageStatus == PoisonMessageStatus.POISON
+ );
+ } catch (final TransientDeliveryException e) {
+ LOG.warn("Transient error processing message id {}, disconnecting.", inboundMessageId, e);
+ offloadedPayloadsToDelete.remove(inboundMessageId);
+ //Disconnect the channel to allow for a reconnect when the HealthCheck passes.
+ disconnectCallback.run();
+ }
+ }
+
+ /**
+ * Handles the logic for injecting taskData from the store into the TaskMessage if required.
+ * Returns true if processing should continue, false if it should stop (e.g. error).
+ * If invalid, handles as poison message (publishes to retry queue) and returns false.
+ */
+ private void handleTaskDataInjection(final TaskMessage taskMessage, final long inboundMessageId,
+ final Optional taskMessageStorageRefOpt)
+ throws InvalidDeliveryException, TransientDeliveryException
+ {
+ final byte[] currentTaskData = taskMessage.getTaskData();
+ final boolean hasStorageRef = taskMessageStorageRefOpt.isPresent();
+ final boolean hasTaskData = currentTaskData != null;
+
+ if (hasTaskData && hasStorageRef) {
+ throw new InvalidDeliveryException(
+ "TaskMessage contains both taskData and a storage reference. This is invalid.", inboundMessageId);
+ }
+ if (!hasTaskData && !hasStorageRef) {
+ throw new InvalidDeliveryException(
+ "TaskMessage contains neither taskData nor a storage reference. This is invalid.", inboundMessageId);
+ }
+ if (hasStorageRef) {
+ final byte[] offloadedTaskData;
+ try {
+ offloadedTaskData = retrieveTaskDataFromStore(taskMessageStorageRefOpt.get(), inboundMessageId);
+ } catch (final ReferenceNotFoundException e) {
+ throw new InvalidDeliveryException(e.getMessage(), inboundMessageId);
+ }
+ taskMessage.setTaskData(offloadedTaskData);
+ }
+ // If hasTaskData and !hasStorageRef, nothing to do
+ }
+
+ private byte[] retrieveTaskDataFromStore(final String taskMessageStorageRef, final long inboundMessageId)
+ throws ReferenceNotFoundException, TransientDeliveryException
+ {
+ try (final var inputStream = dataStore.retrieve(taskMessageStorageRef)) {
+ final var taskData = inputStream.readAllBytes();
+ offloadedPayloadsToDelete.put(inboundMessageId, taskMessageStorageRef);
+ return taskData;
+ } catch (final IOException | DataStoreException ex) {
+ if (ex instanceof ReferenceNotFoundException) {
+ throw (ReferenceNotFoundException)ex;
+ }
+ throw new TransientDeliveryException(
+ "TaskMessage's TaskData could not be retrieved from DataStore", inboundMessageId, ex);
+ }
+ }
+
+ private void handleInvalidDelivery(
+ final long inboundMessageId,
+ final Optional deliveredTaskMessageOpt,
+ final byte[] deliveryMessageData,
+ final Map deliveryHeaders,
+ final String exceptionMesssage
+ )
+ {
+ try {
+ final RabbitTaskInformation taskInformation = new RabbitTaskInformation(String.valueOf(inboundMessageId), true);
+ taskInformation.incrementResponseCount(true);
+ final var publishHeaders = new HashMap<>(deliveryHeaders);
+ publishHeaders.put(RABBIT_HEADER_CAF_WORKER_INVALID, exceptionMesssage);
+
+ publisherEventQueue.add(new WorkerPublishQueueEvent(deliveryMessageData, invalidRoutingKey, taskInformation, publishHeaders));
+
+ if (deliveredTaskMessageOpt.isPresent()) {
+ final var taskMessage = deliveredTaskMessageOpt.get();
+ if(taskMessage.getTracking() != null) {
+ sendFailureTrackingReport(taskMessage, exceptionMesssage, taskInformation);
+ }
+ }
+ } catch (CodecException e) {
+ LOG.error("Failed to serialise report update task data.");
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void sendFailureTrackingReport(
+ final TaskMessage taskMessage,
+ final String invalidDeliveryExceptionMessage,
+ final RabbitTaskInformation rabbitTaskInformation
+ ) throws CodecException {
+ final TrackingReportFailure failure = new TrackingReportFailure();
+ failure.failureId = TaskStatus.INVALID_TASK.toString();
+ failure.failureTime = new Date();
+ failure.failureSource = getWorkerName(taskMessage);
+ failure.failureMessage = invalidDeliveryExceptionMessage;
+
+ final List trackingReports = new ArrayList<>();
+
+ final TrackingReport trackingReport = new TrackingReport();
+ trackingReport.failure = failure;
+ trackingReport.status = TrackingReportStatus.Failed;
+
+ trackingReports.add(trackingReport);
+
+ final TrackingReportTask trackingReportTask = new TrackingReportTask();
+ trackingReportTask.trackingReports = trackingReports;
+
+ final byte[] trackingReportTaskTaskData = codec.serialise(trackingReportTask);
+
+ final TrackingInfo trackingInfo = taskMessage.getTracking();
+
+ final TaskMessage failureReportTaskMessage = new TaskMessage(
+ UUID.randomUUID().toString(), TrackingReportConstants.TRACKING_REPORT_TASK_NAME,
+ TrackingReportConstants.TRACKING_REPORT_TASK_API_VER, trackingReportTaskTaskData, TaskStatus.NEW_TASK,
+ Collections.emptyMap(), trackingInfo.getTrackingPipe(), null, null,
+ taskMessage.getCorrelationId());
+
+ publisherEventQueue.add(new WorkerPublishQueueEvent(codec.serialise(failureReportTaskMessage),
+ trackingInfo.getTrackingPipe(), rabbitTaskInformation, Collections.emptyMap()));
+ }
+
+ private PoisonMessageStatus getPoisonMessageStatus(
+ final boolean isRedelivered,
+ final Map deliveryHeaders,
+ final int retries
+ ) {
+ // If the message is being redelivered it is potentially a poison message.
+ if (isRedelivered) {
+ // If the headers do not contain the delivery count, then it is a classic queue.
+ if (!deliveryHeaders.containsKey(RabbitHeaders.RABBIT_HEADER_CAF_DELIVERY_COUNT)) {
+ // If the retries have not been exceeded, then republish the message
+ // with a header recording the retry count
if (retries < retryLimit) {
- //Republish the delivery with a header recording the incremented number of retries.
- //Classic queues do not record delivery count, so we republish the message with an incremented
- //retry count. This allows us to track the number of attempts to process the message.
- republishClassicRedelivery(delivery, retries);
- return;
+ return PoisonMessageStatus.CLASSIC_POSSIBLY_POISON;
}
- isPoison = true;
- } else {
- isPoison = retries > retryLimit;
}
- } else {
- isPoison = false;
+ return (retries >= retryLimit)
+ ? PoisonMessageStatus.POISON
+ : PoisonMessageStatus.NOT_POISON;
}
+ return PoisonMessageStatus.NOT_POISON;
+ }
- final RabbitTaskInformation taskInformation = new RabbitTaskInformation(String.valueOf(delivery.getEnvelope().getDeliveryTag()), isPoison);
+ private void processDelivery(
+ final long inboundMessageId,
+ final String routingKey,
+ final Map deliveryHeaders,
+ final TaskMessage taskMessage,
+ final byte[] taskMessageByteArray,
+ final boolean isPoison
+ ) {
+ final TrackingInfo trackingInfo = taskMessage.getTracking();
+ final String trackingJobTaskId = trackingInfo != null ? trackingInfo.getJobTaskId() : "untracked";
+ final RabbitTaskInformation taskInformation = new RabbitTaskInformation(
+ String.valueOf(inboundMessageId), isPoison, Optional.of(trackingJobTaskId)
+ );
try {
- LOG.debug("Registering new message {}", taskInformation.getInboundMessageId());
- callback.registerNewTask(taskInformation, delivery.getMessageData(), delivery.getHeaders());
- } catch (InvalidTaskException e) {
- LOG.error("Cannot register new message, rejecting {}", taskInformation.getInboundMessageId(), e);
+ LOG.debug("Registering new message {}", inboundMessageId);
+ callback.registerNewTask(taskInformation, taskMessage, deliveryHeaders);
+ } catch (final InvalidTaskException e) {
+ LOG.error("Cannot register new message, rejecting {}", inboundMessageId, e);
taskInformation.incrementResponseCount(true);
- publisherEventQueue.add(new WorkerPublishQueueEvent(delivery.getMessageData(), retryRoutingKey, taskInformation,
- Collections.singletonMap(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_REJECTED, REJECTED_REASON_TASKMESSAGE)));
- } catch (TaskRejectedException e) {
- LOG.warn("Message {} rejected as a task at this time, returning to queue", taskInformation.getInboundMessageId(), e);
+ final var publishHeaders = new HashMap();
+ publishHeaders.put(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_INVALID, e);
+ publisherEventQueue.add(new WorkerPublishQueueEvent(taskMessageByteArray, invalidRoutingKey, taskInformation, publishHeaders));
+ } catch (final TaskRejectedException e) {
+ LOG.warn("Message {} rejected as a task at this time, returning to queue", inboundMessageId, e);
taskInformation.incrementResponseCount(true);
- publisherEventQueue.add(new WorkerPublishQueueEvent(delivery.getMessageData(), delivery.getEnvelope().getRoutingKey(),
- taskInformation));
+ publisherEventQueue.add(new WorkerPublishQueueEvent(taskMessageByteArray, routingKey, taskInformation, deliveryHeaders));
}
}
@@ -131,6 +364,17 @@ public void processAck(long tag)
LOG.warn("Couldn't ack message {}, will retry", tag, e);
metrics.incremementErrors();
consumerEventQueue.add(new ConsumerAckEvent(tag));
+ return;
+ }
+
+ final String datastorePayloadReference = offloadedPayloadsToDelete.remove(tag);
+ if (datastorePayloadReference != null) {
+ try {
+ dataStore.delete(datastorePayloadReference);
+ } catch (final DataStoreException e) {
+ LOG.warn("Couldn't delete offloaded payload '{}' for delivery tag '{}' from datastore message.",
+ datastorePayloadReference, tag, e);
+ }
}
}
@@ -175,21 +419,59 @@ private void processReject(long id, boolean requeue)
}
}
- /**
- * Republish the delivery to the retry queue with the retry count stamped in the headers.
- *
- * @param delivery the redelivered message
- */
- private void republishClassicRedelivery(final Delivery delivery, final int retries) {
-
- final RabbitTaskInformation taskInformation =
- new RabbitTaskInformation(String.valueOf(delivery.getEnvelope().getDeliveryTag()));
+ private void republishClassicRedelivery(
+ final String deliveryQueue,
+ final long inboundMessageId,
+ final byte[] serializedTaskMessage,
+ final byte[] serializedTaskData,
+ final Map deliveryHeaders,
+ final int retries,
+ final TrackingInfo tracking,
+ final Optional taskMessageStorageRefOpt
+ )
+ {
+ final String trackingJobTaskId = tracking != null ? tracking.getJobTaskId() : "untracked";
+ final RabbitTaskInformation taskInformation = new RabbitTaskInformation(
+ String.valueOf(inboundMessageId), false, Optional.of(trackingJobTaskId));
LOG.debug("Received redelivered message with id {}, retry count {}, retry limit {}, republishing to retry queue",
- delivery.getEnvelope().getDeliveryTag(), retryLimit, retries + 1);
- final Map headers = new HashMap<>();
- headers.put(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_RETRY, String.valueOf(retries + 1));
+ inboundMessageId, retryLimit, retries + 1);
+ final Map publishHeaders = new HashMap<>(deliveryHeaders);
+ publishHeaders.put(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_RETRY, String.valueOf(retries + 1));
taskInformation.incrementResponseCount(true);
- publisherEventQueue.add(new WorkerPublishQueueEvent(delivery.getMessageData(), retryRoutingKey,
- taskInformation, headers));
+ if(taskMessageStorageRefOpt.isPresent()) {
+ if (!retryRoutingKey.equals(deliveryQueue)) {
+ try {
+ final String newStorageReference =
+ dataStore.store(serializedTaskData,
+ taskMessageStorageRefOpt.get().replace(deliveryQueue, retryRoutingKey));
+ publishHeaders.put(RABBIT_HEADER_CAF_PAYLOAD_OFFLOADING_STORAGE_REF, newStorageReference);
+ }
+ catch (final DataStoreException e) {
+ LOG.error("Failed to relocate offloaded payload for message id {} from {} to {}",
+ inboundMessageId, deliveryQueue, retryRoutingKey, e);
+ //Disconnect the channel to allow for a reconnect when the HealthCheck passes.
+ disconnectCallback.run();
+ }
+ }
+ else {
+ //We are reusing the same routing key, so we do not need to relocate the payload.
+ offloadedPayloadsToDelete.remove(inboundMessageId);
+ }
+ }
+ publisherEventQueue.add(new WorkerPublishQueueEvent(serializedTaskMessage, retryRoutingKey, taskInformation, publishHeaders));
+ }
+
+ private String getWorkerName(final TaskMessage taskMessage)
+ {
+ final var taskClassifier = MoreObjects.firstNonNull(taskMessage.getTaskClassifier(), "");
+ if (workerConfiguration != null) {
+ final String workerName = workerConfiguration.getWorkerName();
+
+ if (workerName != null) {
+ return workerName;
+ }
+ }
+
+ return taskClassifier;
}
}
diff --git a/worker-queue-rabbit/src/test/java/com/github/workerframework/queues/rabbit/RabbitWorkerQueueConsumerTest.java b/worker-queue-rabbit/src/test/java/com/github/workerframework/queues/rabbit/RabbitWorkerQueueConsumerTest.java
index 13c39e1f9..fed7f93a9 100644
--- a/worker-queue-rabbit/src/test/java/com/github/workerframework/queues/rabbit/RabbitWorkerQueueConsumerTest.java
+++ b/worker-queue-rabbit/src/test/java/com/github/workerframework/queues/rabbit/RabbitWorkerQueueConsumerTest.java
@@ -15,11 +15,22 @@
*/
package com.github.workerframework.queues.rabbit;
+import com.github.cafapi.common.api.Codec;
+import com.github.cafapi.common.api.CodecException;
+import com.github.cafapi.common.codecs.json.JsonCodec;
+import com.github.workerframework.api.DataStoreException;
import com.github.workerframework.api.InvalidTaskException;
+import com.github.workerframework.api.ManagedDataStore;
import com.github.workerframework.api.TaskCallback;
import com.github.workerframework.api.TaskInformation;
+import com.github.workerframework.api.TaskMessage;
import com.github.workerframework.api.TaskRejectedException;
+import com.github.workerframework.api.TaskStatus;
+import com.github.workerframework.api.TrackingInfo;
+import com.github.workerframework.api.WorkerConfiguration;
import com.github.workerframework.api.WorkerException;
+import com.github.workerframework.datastores.fs.FileSystemDataStore;
+import com.github.workerframework.datastores.fs.FileSystemDataStoreConfiguration;
import com.github.workerframework.util.rabbitmq.ConsumerAckEvent;
import com.github.workerframework.util.rabbitmq.ConsumerDropEvent;
import com.github.workerframework.util.rabbitmq.ConsumerRejectEvent;
@@ -31,15 +42,19 @@
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;
+import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
+import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
@@ -47,24 +62,63 @@
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import static org.mockito.Mockito.mock;
+
public class RabbitWorkerQueueConsumerTest
{
private String testQueue = "testQueue";
- private RabbitTaskInformation taskInformation;
- private byte[] data = "test123".getBytes(StandardCharsets.UTF_8);
+ private RabbitTaskInformation taskInformation;
private Envelope newEnv;
private Envelope poisonEnv;
private Envelope redeliveredEnv;
private String retryKey = "retry";
+ private String invalidKey = "invalid";
private RabbitMetricsReporter metrics = new RabbitMetricsReporter();
- private TaskCallback mockCallback = Mockito.mock(TaskCallback.class);
+ private TaskCallback mockCallback = mock(TaskCallback.class);
+ private File tempDataStore;
+ private ManagedDataStore dataStore;
+ private static Codec codec;
+ private static byte[] data;
+
+ @BeforeClass
+ public static void beforeClass() throws CodecException {
+ codec = new JsonCodec();
+ data = getNewTaskMessage();
+ }
@BeforeMethod
- public void beforeMethod() {
+ public void beforeMethod() throws DataStoreException {
taskInformation = new RabbitTaskInformation("101");
newEnv = new Envelope(Long.valueOf(taskInformation.getInboundMessageId()), false, "", testQueue);
poisonEnv = new Envelope(Long.valueOf(taskInformation.getInboundMessageId()), true, "", testQueue);
redeliveredEnv = new Envelope(Long.valueOf(taskInformation.getInboundMessageId()), true, "", testQueue);
+ tempDataStore = new File("RabbitWorkerQueueConsumerTest");
+ dataStore = new FileSystemDataStore(createConfig());
+ }
+
+ @AfterMethod
+ public void tearDown()
+ {
+ deleteDir(tempDataStore);
+ }
+
+ private void deleteDir(File file)
+ {
+ File[] contents = file.listFiles();
+ if (contents != null) {
+ for (File f : contents) {
+ deleteDir(f);
+ }
+ }
+ file.delete();
+ }
+
+ private FileSystemDataStoreConfiguration createConfig()
+ {
+ final FileSystemDataStoreConfiguration conf = new FileSystemDataStoreConfiguration();
+ conf.setDataDir(tempDataStore.getAbsolutePath());
+ conf.setDataDirHealthcheckTimeoutSeconds(10);
+ return conf;
}
/**
@@ -76,19 +130,21 @@ public void testHandleDelivery()
{
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
- Channel channel = Mockito.mock(Channel.class);
+ Channel channel = mock(Channel.class);
CountDownLatch latch = new CountDownLatch(1);
- TaskCallback callback = Mockito.mock(TaskCallback.class);
+ TaskCallback callback = mock(TaskCallback.class);
Answer a = invocationOnMock -> {
latch.countDown();
return null;
};
Mockito.doAnswer(a).when(callback).registerNewTask(Mockito.any(), Mockito.any(), Mockito.anyMap());
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
- AMQP.BasicProperties prop = Mockito.mock(AMQP.BasicProperties.class);
+ AMQP.BasicProperties prop = mock(AMQP.BasicProperties.class);
Mockito.when(prop.getHeaders()).thenReturn(Collections.emptyMap());
consumer.handleDelivery("consumer", newEnv, prop, data);
Assert.assertTrue(latch.await(1000, TimeUnit.MILLISECONDS));
@@ -105,19 +161,21 @@ public void testPoisonDelivery()
{
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
- Channel channel = Mockito.mock(Channel.class);
+ Channel channel = mock(Channel.class);
CountDownLatch latch = new CountDownLatch(1);
- TaskCallback callback = Mockito.mock(TaskCallback.class);
+ TaskCallback callback = mock(TaskCallback.class);
Answer a = invocationOnMock -> {
latch.countDown();
return null;
};
Mockito.doAnswer(a).when(callback).registerNewTask(Mockito.any(), Mockito.any(), Mockito.anyMap());
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
- AMQP.BasicProperties prop = Mockito.mock(AMQP.BasicProperties.class);
+ AMQP.BasicProperties prop = mock(AMQP.BasicProperties.class);
Map headers = new HashMap<>();
headers.put(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_RETRY, "1");
Mockito.when(prop.getHeaders()).thenReturn(headers);
@@ -131,8 +189,8 @@ public void testPoisonDelivery()
}
/**
- * Send in a new message and verify that if the task registration throws an InvalidTaskException that a new publish request to the
- * reject queue is sent.
+ * Send in a new message and verify that if the task registration throws an InvalidTaskException that a new publish
+ * request to the invalid queue is sent.
*/
@Test
public void testHandleDeliveryInvalid()
@@ -140,28 +198,30 @@ public void testHandleDeliveryInvalid()
{
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
- Channel channel = Mockito.mock(Channel.class);
- TaskCallback callback = Mockito.mock(TaskCallback.class);
+ Channel channel = mock(Channel.class);
+ TaskCallback callback = mock(TaskCallback.class);
Answer a = invocationOnMock -> {
throw new InvalidTaskException("blah");
};
Mockito.doAnswer(a).when(callback).registerNewTask(Mockito.any(), Mockito.any(), Mockito.anyMap());
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
- AMQP.BasicProperties prop = Mockito.mock(AMQP.BasicProperties.class);
+ AMQP.BasicProperties prop = mock(AMQP.BasicProperties.class);
Mockito.when(prop.getHeaders()).thenReturn(Collections.emptyMap());
consumer.handleDelivery("consumer", newEnv, prop, data);
Event pubEvent = publisherEvents.poll(1, TimeUnit.SECONDS);
Assert.assertNotNull(pubEvent);
- WorkerPublisher publisher = Mockito.mock(WorkerPublisher.class);
+ WorkerPublisher publisher = mock(WorkerPublisher.class);
ArgumentCaptor> captor = buildStringObjectMapCaptor();
pubEvent.handleEvent(publisher);
- Mockito.verify(publisher, Mockito.times(1)).handlePublish(Mockito.eq(data), Mockito.eq(retryKey), Mockito.any(RabbitTaskInformation.class), captor.capture());
- Assert.assertTrue(captor.getValue().containsKey(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_REJECTED));
- Assert.assertEquals(WorkerQueueConsumerImpl.REJECTED_REASON_TASKMESSAGE,
- captor.getValue().get(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_REJECTED));
+ Mockito.verify(publisher, Mockito.times(1)).handlePublish(Mockito.eq(data), Mockito.eq(invalidKey), Mockito.any(RabbitTaskInformation.class), captor.capture());
+ Assert.assertTrue(captor.getValue().containsKey(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_INVALID));
+ Assert.assertEquals(captor.getValue().get(RabbitHeaders.RABBIT_HEADER_CAF_WORKER_INVALID).toString(),
+ "com.github.workerframework.api.InvalidTaskException: blah");
consumer.shutdown();
}
@@ -175,22 +235,24 @@ public void testHandleDeliveryRejected()
{
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
- Channel channel = Mockito.mock(Channel.class);
- TaskCallback callback = Mockito.mock(TaskCallback.class);
+ Channel channel = mock(Channel.class);
+ TaskCallback callback = mock(TaskCallback.class);
Answer a = invocationOnMock -> {
throw new TaskRejectedException("blah");
};
Mockito.doAnswer(a).when(callback).registerNewTask(Mockito.any(), Mockito.any(), Mockito.anyMap());
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
- AMQP.BasicProperties prop = Mockito.mock(AMQP.BasicProperties.class);
+ AMQP.BasicProperties prop = mock(AMQP.BasicProperties.class);
Mockito.when(prop.getHeaders()).thenReturn(Collections.emptyMap());
consumer.handleDelivery("consumer", newEnv, prop, data);
Event pubEvent = publisherEvents.poll(1, TimeUnit.SECONDS);
Assert.assertNotNull(pubEvent);
- WorkerPublisher publisher = Mockito.mock(WorkerPublisher.class);
+ WorkerPublisher publisher = mock(WorkerPublisher.class);
ArgumentCaptor> captor = buildStringObjectMapCaptor();
pubEvent.handleEvent(publisher);
Mockito.verify(publisher, Mockito.times(1)).handlePublish(Mockito.eq(data), Mockito.eq(testQueue), Mockito.any(RabbitTaskInformation.class), captor.capture());
@@ -208,18 +270,20 @@ public void testHandleRedelivery()
{
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
- Channel channel = Mockito.mock(Channel.class);
- TaskCallback callback = Mockito.mock(TaskCallback.class);
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ Channel channel = mock(Channel.class);
+ TaskCallback callback = mock(TaskCallback.class);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ callback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
- AMQP.BasicProperties prop = Mockito.mock(AMQP.BasicProperties.class);
+ AMQP.BasicProperties prop = mock(AMQP.BasicProperties.class);
Mockito.when(prop.getHeaders()).thenReturn(Collections.emptyMap());
consumer.handleDelivery("consumer", redeliveredEnv, prop, data);
Event pubEvent = publisherEvents.poll(1, TimeUnit.SECONDS);
Assert.assertNotNull(pubEvent);
- WorkerPublisher publisher = Mockito.mock(WorkerPublisher.class);
+ WorkerPublisher publisher = mock(WorkerPublisher.class);
ArgumentCaptor> captor = buildStringObjectMapCaptor();
pubEvent.handleEvent(publisher);
Mockito.verify(publisher, Mockito.times(1)).handlePublish(Mockito.eq(data), Mockito.eq(retryKey), Mockito.any(RabbitTaskInformation.class), captor.capture());
@@ -238,13 +302,15 @@ public void testHandleDeliveryAck()
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
CountDownLatch channelLatch = new CountDownLatch(1);
- Channel channel = Mockito.mock(Channel.class);
+ Channel channel = mock(Channel.class);
Answer a = invocationOnMock -> {
channelLatch.countDown();
return null;
};
Mockito.doAnswer(a).when(channel).basicAck(Mockito.eq(Long.valueOf(taskInformation.getInboundMessageId())), Mockito.anyBoolean());
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(mockCallback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ mockCallback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
@@ -263,13 +329,15 @@ public void testHandleDeliveryReject()
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
CountDownLatch channelLatch = new CountDownLatch(1);
- Channel channel = Mockito.mock(Channel.class);
+ Channel channel = mock(Channel.class);
Answer a = invocationOnMock -> {
channelLatch.countDown();
return null;
};
Mockito.doAnswer(a).when(channel).basicReject(Mockito.eq(Long.valueOf(taskInformation.getInboundMessageId())), Mockito.eq(true));
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(mockCallback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ mockCallback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
@@ -288,13 +356,15 @@ public void testHandleDeliveryDrop()
BlockingQueue> consumerEvents = new LinkedBlockingQueue<>();
BlockingQueue> publisherEvents = new LinkedBlockingQueue<>();
CountDownLatch channelLatch = new CountDownLatch(1);
- Channel channel = Mockito.mock(Channel.class);
+ Channel channel = mock(Channel.class);
Answer a = invocationOnMock -> {
channelLatch.countDown();
return null;
};
Mockito.doAnswer(a).when(channel).basicReject(Long.valueOf(taskInformation.getInboundMessageId()), false);
- WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(mockCallback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1);
+ WorkerQueueConsumerImpl impl = new WorkerQueueConsumerImpl(
+ mockCallback, metrics, consumerEvents, channel, publisherEvents, retryKey, 1, invalidKey,
+ dataStore, codec, () -> {}, mock(WorkerConfiguration.class));
DefaultRabbitConsumer consumer = new DefaultRabbitConsumer(consumerEvents, impl);
Thread t = new Thread(consumer);
t.start();
@@ -308,4 +378,18 @@ private static ArgumentCaptor> buildStringObjectMapCaptor()
{
return ArgumentCaptor.forClass(Map.class);
}
+
+ private static byte[] getNewTaskMessage() throws CodecException {
+ final var trackingInfo = new TrackingInfo("task1", new Date(), 1, "http://hello.com", "pipe", "to");
+ return codec.serialise(new TaskMessage(
+ "task1",
+ "ACTUAL_CLASSIFIER",
+ 1,
+ "test123".getBytes(StandardCharsets.UTF_8),
+ TaskStatus.NEW_TASK,
+ new HashMap<>(),
+ "to",
+ trackingInfo
+ ));
+ }
}
diff --git a/worker-store-fs/pom.xml b/worker-store-fs/pom.xml
index ea8ae1ee5..c2ff6b3cf 100644
--- a/worker-store-fs/pom.xml
+++ b/worker-store-fs/pom.xml
@@ -22,12 +22,12 @@
4.0.0
worker-store-fs
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
com.github.workerframework
worker-framework-aggregator
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
diff --git a/worker-store-http/pom.xml b/worker-store-http/pom.xml
index 6ac9543c9..4a7a30876 100644
--- a/worker-store-http/pom.xml
+++ b/worker-store-http/pom.xml
@@ -22,12 +22,12 @@
4.0.0
worker-store-http
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
com.github.workerframework
worker-framework-aggregator
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
diff --git a/worker-store-mem/pom.xml b/worker-store-mem/pom.xml
index 0023f3c0d..d25553782 100644
--- a/worker-store-mem/pom.xml
+++ b/worker-store-mem/pom.xml
@@ -22,12 +22,12 @@
4.0.0
worker-store-mem
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
com.github.workerframework
worker-framework-aggregator
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
diff --git a/worker-store-s3/pom.xml b/worker-store-s3/pom.xml
index 96036adac..841b32479 100644
--- a/worker-store-s3/pom.xml
+++ b/worker-store-s3/pom.xml
@@ -22,12 +22,12 @@
4.0.0
worker-store-s3
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
com.github.workerframework
worker-framework-aggregator
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
diff --git a/worker-test/pom.xml b/worker-test/pom.xml
index 238fb5b1b..51ac51e87 100644
--- a/worker-test/pom.xml
+++ b/worker-test/pom.xml
@@ -24,7 +24,7 @@
com.github.workerframework
worker-framework-aggregator
- 9.1.2-SNAPSHOT
+ 10.0.0-SNAPSHOT
worker-test
@@ -88,6 +88,29 @@
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+
+
+ integration-test
+ integration-test
+
+ integration-test
+
+
+
+ http://${docker.host.address}:${webdav.port}/webdav
+ ${docker.host.address}
+ ${rabbitmq.node.port}
+ guest
+ guest
+
+
+
+
+
+
org.apache.maven.plugins
@@ -250,7 +273,7 @@
- worker-test
+ PayloadOffloadingIT
${targetDockerRegistryPath}/worker-test:${project.version}
${projectDockerRegistry}/cafapi/opensuse-jre17
@@ -295,11 +318,110 @@
linux/amd64
- ${worker.adminport}:8081
- ${worker.debugport}:5005
+ ${PayloadOffloadingIT.httpport}:8080
+ ${PayloadOffloadingIT.adminport}:8081
+ ${PayloadOffloadingIT.debugport}:5005
+
+
+ PayloadOffloadingIT-in
+ PayloadOffloadingIT-out
+ PayloadOffloadingIT-reject
+ PayloadOffloadingIT-invalid
+ /srv/common/webdav
+ true
+ 1
+ queues
+
+ -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
+
+ 1
+ amqps
+ 5671
+ /test-keystore
+ ca_certificate.pem
+
+
+
+ webdav
+ keystore
+
+
+
+ rabbitmq
+
+
+ true
+
+
+
+ http://${docker.host.address}:${PayloadOffloadingIT.adminport}
+
+
+ 500
+
+
+
+
+ PoisonMessageIT-1
+ ${targetDockerRegistryPath}/worker-test:${project.version}
+
+ linux/amd64
+
+ ${PoisonMessageIT-1.httpport}:8080
+ ${PoisonMessageIT-1.adminport}:8081
+ ${PoisonMessageIT-1.debugport}:5005
+
+
+ PoisonMessageIT-in
+ PoisonMessageIT-out
+ PoisonMessageIT-reject
+ PoisonMessageIT-invalid
+ /srv/common/webdav
+
+ -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
+
+ 1
+ amqps
+ 5671
+ /test-keystore
+ ca_certificate.pem
+
+
+
+ webdav
+ keystore
+
+
+
+ rabbitmq
+
+
+ true
+
+
+
+ http://${docker.host.address}:${PoisonMessageIT-1.adminport}
+
+
+ 500
+
+
+
+
+ PoisonMessageIT-2
+ ${targetDockerRegistryPath}/worker-test:${project.version}
+
+ linux/amd64
+
+ ${PoisonMessageIT-2.httpport}:8080
+ ${PoisonMessageIT-2.adminport}:8081
+ ${PoisonMessageIT-2.debugport}:5005
- 10
+ PoisonMessageIT-in
+ PoisonMessageIT-out
+ PoisonMessageIT-reject
+ PoisonMessageIT-invalid
/srv/common/webdav
-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
@@ -324,25 +446,34 @@
- http://${docker.host.address}:${worker.adminport}
+ http://${docker.host.address}:${PoisonMessageIT-2.adminport}
500
+
+
- worker-test-2
+ PoisonMessageIT-Offloading-1
${targetDockerRegistryPath}/worker-test:${project.version}
linux/amd64
- ${worker.testhttpport2}:8080
- ${worker.testadminport2}:8081
- ${worker.testdebugport2}:5005
+ ${PoisonMessageIT-Offloading-1.httpport}:8080
+ ${PoisonMessageIT-Offloading-1.adminport}:8081
+ ${PoisonMessageIT-Offloading-1.debugport}:5005
+ PoisonMessageIT-Offloading-in
+ PoisonMessageIT-Offloading-out
+ PoisonMessageIT-Offloading-reject
+ PoisonMessageIT-Offloading-invalid
/srv/common/webdav
+ true
+ 1
+ queues
-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
@@ -366,13 +497,114 @@
- http://${docker.host.address}:${worker.testadminport2}
+ http://${docker.host.address}:${PoisonMessageIT-Offloading-1.adminport}
500
+
+ PoisonMessageIT-Offloading-2
+ ${targetDockerRegistryPath}/worker-test:${project.version}
+
+ linux/amd64
+
+ ${PoisonMessageIT-Offloading-2.httpport}:8080
+ ${PoisonMessageIT-Offloading-2.adminport}:8081
+ ${PoisonMessageIT-Offloading-2.debugport}:5005
+
+
+ PoisonMessageIT-Offloading-in
+ PoisonMessageIT-Offloading-out
+ PoisonMessageIT-Offloading-reject
+ PoisonMessageIT-Offloading-invalid
+ /srv/common/webdav
+ true
+ 1
+ queues
+
+ -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
+
+ 1
+ amqps
+ 5671
+ /test-keystore
+ ca_certificate.pem
+
+
+
+ webdav
+ keystore
+
+
+
+ rabbitmq
+
+
+ true
+
+
+
+ http://${docker.host.address}:${PoisonMessageIT-Offloading-2.adminport}
+
+
+ 500
+
+
+
+
+
+
+ PayloadOffloadingIT-Terminal
+ ${targetDockerRegistryPath}/worker-test:${project.version}
+
+ linux/amd64
+
+ ${PayloadOffloadingIT-Terminal.httpport}:8080
+ ${PayloadOffloadingIT-Terminal.adminport}:8081
+ ${PayloadOffloadingIT-Terminal.debugport}:5005
+
+
+ PayloadOffloadingIT-Terminal-in
+ PayloadOffloadingIT-Terminal-out
+ PayloadOffloadingIT-Terminal-reject
+ PayloadOffloadingIT-Terminal-invalid
+ /srv/common/webdav
+ true
+ 1
+ queues
+
+ -agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=*:5005
+
+ 1
+ amqps
+ 5671
+ /test-keystore
+ ca_certificate.pem
+
+
+
+ webdav
+ keystore
+
+
+
+ rabbitmq
+
+
+ true
+
+
+
+ http://${docker.host.address}:${PayloadOffloadingIT-Terminal.adminport}
+
+
+ 500
+
+
+
+
worker-test-no-valid-cert
@@ -380,9 +612,9 @@
linux/amd64
- ${worker.testhttpport3}:8080
- ${worker.testadminport3}:8081
- ${worker.testdebugport3}:5005
+ ${worker-test-no-valid-cert.httpport}:8080
+ ${worker-test-no-valid-cert.adminport}:8081
+ ${worker-test-no-valid-cert.debugport}:5005
/srv/common/webdav
@@ -407,7 +639,7 @@
- http://${docker.host.address}:${worker.testhttpport3}/health-check?type=READY
+ http://${docker.host.address}:${worker-test-no-valid-cert.httpport}/health-check?type=READY
503
@@ -428,14 +660,38 @@
15672
25672
9090
- 8081
- 5005
- 8084
- 8082
- 8083
- 8085
- 5006
- 5007
+
+ 8080
+ 8081
+ 5010
+
+ 8082
+ 8083
+ 5011
+
+ 8084
+ 8085
+ 5012
+
+ 8086
+ 8087
+ 5013
+
+ 8088
+ 8089
+ 5014
+
+ 8090
+ 8091
+ 5015
+
+ 8092
+ 8093
+ 5016
+
+ 8094
+ 8095
+ 5017