WorkerFramework · DavidGerald-Milligan · Apr 10, 2025 · Apr 11, 2025 · Apr 16, 2025 · Apr 16, 2025
diff --git a/docs/pom.xml b/docs/pom.xml
@@ -28,7 +28,7 @@
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <properties>

diff --git a/pom.xml b/pom.xml
@@ -23,7 +23,7 @@
 
     <groupId>com.github.workerframework</groupId>
     <artifactId>worker-framework-aggregator</artifactId>
-    <version>9.1.2-SNAPSHOT</version>
+    <version>10.0.0-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <name>Worker Framework</name>
@@ -273,52 +273,52 @@
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>standard-worker-container</artifactId>
                 <type>pom</type>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>util-rabbitmq</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-api</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-caf</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-configs</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-core</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-default-configs</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-queue-rabbit</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-store-fs</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.github.workerframework</groupId>
                 <artifactId>worker-tracking-report</artifactId>
-                <version>9.1.2-SNAPSHOT</version>
+                <version>10.0.0-SNAPSHOT</version>
             </dependency>
             <dependency>
                 <groupId>com.google.code.findbugs</groupId>

diff --git a/release-notes-10.0.0.md b/release-notes-10.0.0.md
@@ -0,0 +1,14 @@
+#### Version Number
+${version-number}
+
+#### Breaking Changes
+- US1009117: Remove reliance on large messages being supported by the underlying queue provider.
+  - The TaskCallback interface has been updated to expect a TaskMessage in place of a byte array.
+  - The WorkerQueueProvider interface has been updated to expect a ManagedDataStore for storing large messages
+    and a Codec for serialization/deserialization of messages prior to storage/retrieval from the datastore.
+
+#### New Features
+- None    
+
+#### Known Issues
+- None
diff --git a/release-notes-9.1.2.md b/release-notes-9.1.2.md
diff --git a/standard-worker-container/pom.xml b/standard-worker-container/pom.xml
@@ -28,7 +28,7 @@
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/util-rabbitmq/pom.xml b/util-rabbitmq/pom.xml
@@ -22,12 +22,12 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>util-rabbitmq</artifactId>
-    <version>9.1.2-SNAPSHOT</version>
+    <version>10.0.0-SNAPSHOT</version>
 
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/util-rabbitmq/src/main/java/com/github/workerframework/util/rabbitmq/RabbitHeaders.java b/util-rabbitmq/src/main/java/com/github/workerframework/util/rabbitmq/RabbitHeaders.java
@@ -21,6 +21,8 @@
 public class RabbitHeaders
 {
     public static final String RABBIT_HEADER_CAF_WORKER_REJECTED = "x-caf-worker-rejected";
+    public static final String RABBIT_HEADER_CAF_WORKER_INVALID = "x-caf-worker-invalid";
     public static final String RABBIT_HEADER_CAF_WORKER_RETRY = "x-caf-worker-retry";
     public static final String RABBIT_HEADER_CAF_DELIVERY_COUNT = "x-delivery-count";
+    public static final String RABBIT_HEADER_CAF_PAYLOAD_OFFLOADING_STORAGE_REF = "x-caf-payload-offloading-storage-ref";
 }
diff --git a/worker-api/pom.xml b/worker-api/pom.xml
@@ -22,12 +22,12 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>worker-api</artifactId>
-    <version>9.1.2-SNAPSHOT</version>
+    <version>10.0.0-SNAPSHOT</version>
 
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/worker-api/src/main/java/com/github/workerframework/api/TaskCallback.java b/worker-api/src/main/java/com/github/workerframework/api/TaskCallback.java
@@ -27,12 +27,12 @@ public interface TaskCallback
      * Announce to the worker core that a new task has been picked off the queue for processing.
      *
      * @param taskInformation contains an arbitrary task reference
-     * @param taskData the task data that is specific to the workers hosted
+     * @param taskMessage the task data that is specific to the workers hosted
      * @param headers the map of key/value paired headers on the message
      * @throws TaskRejectedException if the worker framework rejected execution of the task at this time
      * @throws InvalidTaskException if the worker framework indicates this task is invalid and cannot possibly be executed
      */
-    void registerNewTask(TaskInformation taskInformation, byte[] taskData, Map<String, Object> headers)
+    void registerNewTask(TaskInformation taskInformation, TaskMessage taskMessage, Map<String, Object> headers)
         throws TaskRejectedException, InvalidTaskException;
 
     /**

diff --git a/worker-api/src/main/java/com/github/workerframework/api/TaskMessage.java b/worker-api/src/main/java/com/github/workerframework/api/TaskMessage.java
@@ -53,7 +53,6 @@ public final class TaskMessage
     /**
      * The serialised data of the task-specific message.
      */
-    @NotNull
     private byte[] taskData;
 
     /**
@@ -129,7 +128,7 @@ public TaskMessage(final String taskId, final String taskClassifier, final int t
         this.taskId = Objects.requireNonNull(taskId);
         this.taskClassifier = Objects.requireNonNull(taskClassifier);
         this.taskApiVersion = Objects.requireNonNull(taskApiVersion);
-        this.taskData = Objects.requireNonNull(taskData);
+        this.taskData = taskData;
         this.taskStatus = Objects.requireNonNull(taskStatus);
         this.context = Objects.requireNonNull(context);
         this.to = to;

diff --git a/worker-api/src/main/java/com/github/workerframework/api/TaskRejectedException.java b/worker-api/src/main/java/com/github/workerframework/api/TaskRejectedException.java
@@ -17,6 +17,8 @@
 
 /**
  * Indicates that a task cannot be accepted right now, but that it should be retried at a later time.
+ * This exception does not result in the message being placed on the reject queue.
+ * The reject queue is only used for tasks that cannot be processed at all and have been identified as poisonous.
  */
 public class TaskRejectedException extends WorkerException
 {

diff --git a/worker-api/src/main/java/com/github/workerframework/api/WorkerQueue.java b/worker-api/src/main/java/com/github/workerframework/api/WorkerQueue.java
@@ -32,7 +32,7 @@ public interface WorkerQueue
      * @param isLastMessage the boolean to indicate if current message is final message for the task
      * @throws QueueException if the message cannot be submitted
      */
-    void publish(TaskInformation taskInformation, byte[] taskMessage, String targetQueue, Map<String, Object> headers,
+    void publish(TaskInformation taskInformation, TaskMessage taskMessage, String targetQueue, Map<String, Object> headers,
                  boolean isLastMessage) throws QueueException;
 
     /**
@@ -45,7 +45,7 @@ void publish(TaskInformation taskInformation, byte[] taskMessage, String targetQ
      * @throws QueueException if the message cannot be submitted
      */
 
-    void publish(TaskInformation taskInformation, byte[] taskMessage, String targetQueue, Map<String, Object> headers)
+    void publish(TaskInformation taskInformation, TaskMessage taskMessage, String targetQueue, Map<String, Object> headers)
         throws QueueException;
 
     /**

diff --git a/worker-api/src/main/java/com/github/workerframework/api/WorkerQueueProvider.java b/worker-api/src/main/java/com/github/workerframework/api/WorkerQueueProvider.java
@@ -15,6 +15,7 @@
  */
 package com.github.workerframework.api;
 
+import com.github.cafapi.common.api.Codec;
 import com.github.cafapi.common.api.ConfigurationSource;
 
 /**
@@ -26,10 +27,15 @@ public interface WorkerQueueProvider
      * Create a new WorkerQueue instance.
      *
      * @param configurationSource used for configuring the WorkerQueue
-     * @param maxTasks the maximum number of tasks the worker can perform at once
+     * @param maxTasks            the maximum number of tasks the worker can perform at once
+     * @param invalidQueue        the queue in which to place tasks that are invalid or cannot be processed
+     * @param dataStore           the managed data store that the worker will use to store data that exceeds a threshold.
+     * @param codec               the codec used for serialization deserialization of data.
+     * @param workerConfiguration
      * @return a new WorkerQueue instance
      * @throws QueueException if a WorkerQueue could not be created
      */
-    ManagedWorkerQueue getWorkerQueue(ConfigurationSource configurationSource, int maxTasks)
+    ManagedWorkerQueue getWorkerQueue(ConfigurationSource configurationSource, int maxTasks, String invalidQueue,
+                                      ManagedDataStore dataStore, Codec codec, WorkerConfiguration workerConfiguration)
         throws QueueException;
 }
diff --git a/worker-caf/pom.xml b/worker-caf/pom.xml
@@ -22,12 +22,12 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>worker-caf</artifactId>
-    <version>9.1.2-SNAPSHOT</version>
+    <version>10.0.0-SNAPSHOT</version>
 
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/worker-configs/pom.xml b/worker-configs/pom.xml
@@ -22,12 +22,12 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>worker-configs</artifactId>
-    <version>9.1.2-SNAPSHOT</version>
+    <version>10.0.0-SNAPSHOT</version>
 
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/worker-core/pom.xml b/worker-core/pom.xml
@@ -22,12 +22,12 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>worker-core</artifactId>
-    <version>9.1.2-SNAPSHOT</version>
+    <version>10.0.0-SNAPSHOT</version>
 
     <parent>
         <groupId>com.github.workerframework</groupId>
         <artifactId>worker-framework-aggregator</artifactId>
-        <version>9.1.2-SNAPSHOT</version>
+        <version>10.0.0-SNAPSHOT</version>
     </parent>
 
     <dependencies>

diff --git a/worker-core/readme.md b/worker-core/readme.md
@@ -266,7 +266,6 @@ the current input queue. Default is True.
   indicated as requeued by the WorkerQueue
  - core.currentIdleTime: the time in milliseconds since the worker was doing
   anything useful.
- - core.inputSizes: histogram of input (task) message sizes in bytes
  - core.outputSize: histogram of output (result) messages sizes in bytes
  - config.lookups: the number of configuration lookups performed by the
   ConfigurationSource.

diff --git a/worker-core/src/main/java/com/github/workerframework/core/WorkerApplication.java b/worker-core/src/main/java/com/github/workerframework/core/WorkerApplication.java
@@ -125,7 +125,8 @@ public void run(final WorkerConfiguration workerConfiguration, final Environment
         WorkerFactory workerFactory = workerProvider.getWorkerFactory(config, store, codec);
         WorkerThreadPool wtp = WorkerThreadPool.create(workerFactory);
         final int nThreads = workerFactory.getWorkerThreads();
-        ManagedWorkerQueue workerQueue = queueProvider.getWorkerQueue(config, nThreads);
+        ManagedWorkerQueue workerQueue = queueProvider.getWorkerQueue(config, nThreads, workerFactory.getInvalidTaskQueue(), store, codec,
+            workerFactory.getWorkerConfiguration());
         TransientHealthCheck transientHealthCheck = new TransientHealthCheck();
         WorkerCore core = new WorkerCore(codec, wtp, workerQueue, workerFactory, path, environment.healthChecks(), transientHealthCheck);
         HealthConfiguration healthConfiguration = config.getConfiguration(HealthConfiguration.class);