Dw 5.x update

.gitignore

@@ -25,6 +25,9 @@ hs_err_pid*
 *.ipr
 *.iml
 venv
+.project
+.factorypath
+.classpath
 
 ## Plugin-specific files:
 
@@ -90,3 +93,9 @@ out.*
 */creds.sh
 .run
 build*
+
+# Agents
+.pi/
+.planning/
+.settings/
+config.json

memq-actor/pom.xml

@@ -5,17 +5,13 @@
     <parent>
         <groupId>io.appform.memq</groupId>
         <artifactId>memq</artifactId>
-        <version>0.1.8</version>
+        <version>5.0.2-0-1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>memq-actor</artifactId>
     <name>MemQ Core Actors Library</name>
 
-    <properties>
-        <maven.compiler.source>17</maven.compiler.source>
-        <maven.compiler.target>17</maven.compiler.target>
-    </properties>
 
     <dependencies>
         <dependency>
@@ -26,12 +22,15 @@
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
-            <version>${slf4j-api.version}</version>
         </dependency>
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>
-            <version>${guava.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>jakarta.validation</groupId>
+            <artifactId>jakarta.validation-api</artifactId>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
@@ -41,23 +40,18 @@
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
-            <version>2.16.1</version>
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.16.1</version>
         </dependency>
         <dependency>
             <groupId>io.dropwizard.metrics</groupId>
             <artifactId>metrics-core</artifactId>
-            <version>${metrics-core.version}</version>
         </dependency>
-
         <dependency>
             <groupId>ch.qos.logback</groupId>
             <artifactId>logback-classic</artifactId>
-            <version>1.2.13</version>
             <scope>test</scope>
         </dependency>
 

memq-actor/src/main/java/io/appform/memq/HighLevelActorConfig.java

@@ -4,13 +4,13 @@
 import io.appform.memq.exceptionhandler.config.ExceptionHandlerConfig;
 import io.appform.memq.retry.config.NoRetryConfig;
 import io.appform.memq.retry.config.RetryConfig;
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.Max;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotNull;
 import lombok.*;
 import lombok.extern.jackson.Jacksonized;
 
-import javax.validation.Valid;
-import javax.validation.constraints.Max;
-import javax.validation.constraints.Min;
-import javax.validation.constraints.NotNull;
 
 @Value
 @Builder

memq-actor/src/main/java/io/appform/memq/actor/Actor.java

@@ -1,6 +1,5 @@
 package io.appform.memq.actor;
 
-import com.google.common.base.Preconditions;
 import io.appform.memq.observer.ActorObserver;
 import io.appform.memq.observer.ActorObserverContext;
 import io.appform.memq.observer.ObserverMessageMeta;
@@ -64,9 +63,12 @@ public Actor(
         // Switch ensures validation is performed for all dispatcher types.
         // Adding a new DispatcherType will cause a compile error here if not handled.
         switch (dispatcherType){
-            case SYNC ->
-                    Preconditions.checkArgument( maxConcurrencyPerPartition == maxSizePerPartition,
-                    "Max Queue size and max concurrency has to be same for sync dispatcher");
+            case SYNC -> {
+                if (maxConcurrencyPerPartition != maxSizePerPartition) {
+                    throw new IllegalArgumentException(
+                            "Max Queue size and max concurrency has to be same for sync dispatcher");
+                }
+            }
             case ASYNC_ISOLATED -> {
                 // No-op: ASYNC_ISOLATED dispatcher has no specific validation requirements
             }
@@ -136,11 +138,11 @@ public final void close() {
     }
 
    void processWithObserver(final InternalMessage<M> internalMessage) {
-        val observerMessageMeta = new ObserverMessageMeta(internalMessage.getId(), internalMessage.getPublishedAt(),
-                internalMessage.getValidTill());
+        val observerMessageMeta = new ObserverMessageMeta(internalMessage.id(), internalMessage.publishedAt(),
+                internalMessage.validTill());
         this.rootObserver.execute(ActorObserverContext.builder()
                         .messageMeta(observerMessageMeta)
-                        .message(internalMessage.getMessage())
+                        .message(internalMessage.message())
                         .operation(ActorOperation.CONSUME)
                         .actorName(this.name)
                         .build(),
@@ -160,12 +162,12 @@ Dispatcher<M> getMessageDispatcher() {
     }
 
     private boolean process(final InternalMessage<M> internalMessage) {
-        val id = internalMessage.getId();
-        val message = internalMessage.getMessage();
+        val id = internalMessage.id();
+        val message = internalMessage.message();
         var status = false;
-        var messageMeta = new MessageMeta(internalMessage.getPublishedAt(),
-                internalMessage.getValidTill(),
-                internalMessage.getHeaders());
+        var messageMeta = new MessageMeta(internalMessage.publishedAt(),
+                internalMessage.validTill(),
+                internalMessage.headers());
         try {
             val valid = this.rootObserver.execute(ActorObserverContext.builder()
                             .message(message)
@@ -177,23 +179,21 @@ private boolean process(final InternalMessage<M> internalMessage) {
                 log.debug("Message validation failed for message: {}", message);
                 return false;
             }
-            else {
-                status = this.retryer.execute(() -> {
-                    messageMeta.incrementAttempt();
-                    return this.consumerHandler.test(message, messageMeta);
-                });
-                if (!status) {
-                    log.debug("Consumer failed for message: {}", message);
-                    this.rootObserver.execute(ActorObserverContext.builder()
-                                    .message(message)
-                                    .operation(ActorOperation.SIDELINE)
-                                    .actorName(this.name)
-                                    .build(),
-                            () -> {
-                                this.sidelineHandler.accept(message, messageMeta);
-                                return true;
-                            });
-                }
+            status = this.retryer.execute(() -> {
+                messageMeta.incrementAttempt();
+                return this.consumerHandler.test(message, messageMeta);
+            });
+            if (!status) {
+                log.debug("Consumer failed for message: {}", message);
+                this.rootObserver.execute(ActorObserverContext.builder()
+                                .message(message)
+                                .operation(ActorOperation.SIDELINE)
+                                .actorName(this.name)
+                                .build(),
+                        () -> {
+                            this.sidelineHandler.accept(message, messageMeta);
+                            return true;
+                        });
             }
         } catch (Exception e) {
             log.error("Error processing message : " + id, e);

memq-actor/src/main/java/io/appform/memq/actor/AsyncIsolatedThreadpoolDispatcher.java

@@ -27,9 +27,9 @@ public final void register(final Mailbox<M> inMailbox) {
     //Always executed inside mailbox lock
     @Override
     public final void deRegister(final Mailbox<M> inMailbox) {
-        if(registeredMailboxWorker.containsKey(inMailbox.getPartition())) {
-            registeredMailboxWorker.get(inMailbox.getPartition()).close();
-            registeredMailboxWorker.remove(inMailbox.getPartition());
+        val worker = registeredMailboxWorker.remove(inMailbox.getPartition());
+        if (worker != null) {
+            worker.close();
         }
     }
 

memq-actor/src/main/java/io/appform/memq/actor/Dispatcher.java

@@ -1,12 +1,11 @@
 package io.appform.memq.actor;
 
-import com.google.common.collect.Sets;
 import lombok.val;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Set;
-import java.util.stream.Collectors;
+import java.util.ArrayList;
+import java.util.List;
 
 interface Dispatcher<M extends Message> extends AutoCloseable {
 
@@ -20,14 +19,23 @@ interface Dispatcher<M extends Message> extends AutoCloseable {
 
     //Always executed inside mailbox lock
     default void dispatch(final Mailbox<M> mailbox) {
-        //Find new messages
-        val newInOrderedMessages = mailbox.getMessages().keySet()
-                .stream()
-                .limit(mailbox.getMaxConcurrency())
-                .collect(Collectors.toSet());
-        val newMessageIds = Set.copyOf(Sets.difference(newInOrderedMessages, mailbox.getInFlight()));
-        if (newMessageIds.isEmpty()) {
-            if(mailbox.getInFlight().size() == mailbox.getMaxConcurrency()) {
+        val inFlight = mailbox.getInFlight();
+        //Find new messages, respecting insertion order and max concurrency
+        int considered = 0;
+        List<InternalMessage<M>> newMessages = new ArrayList<>(mailbox.getMaxConcurrency());
+        for (val entry : mailbox.getMessages().entrySet()) {
+            if (considered >= mailbox.getMaxConcurrency()) {
+                break;
+            }
+            considered++;
+            val id = entry.getKey();
+            if (!inFlight.contains(id)) {
+                inFlight.add(id);
+                newMessages.add(entry.getValue());
+            }
+        }
+        if (newMessages.isEmpty()) {
+            if (inFlight.size() == mailbox.getMaxConcurrency()) {
                 log.warn("Reached max concurrency:{}. Ignoring consumption till inflight messages are consumed",
                         mailbox.getMaxConcurrency());
             }
@@ -36,12 +44,12 @@ default void dispatch(final Mailbox<M> mailbox) {
             }
             return;
         }
-        mailbox.getInFlight().addAll(newMessageIds);
-        val messagesToBeDelivered = newMessageIds.stream()
-                .map(mailbox.getMessages()::get)
-                .toList();
-        messagesToBeDelivered.forEach(internalMessage -> mailbox.getActor().getExecutorService().submit(() -> {
-            val id = internalMessage.getId();
+        dispatchNew(mailbox, newMessages);
+    }
+
+    default void dispatchNew(final Mailbox<M> mailbox, List<InternalMessage<M>> newMessages) {
+        List.copyOf(newMessages).forEach(internalMessage -> mailbox.getActor().getExecutorService().submit(() -> {
+            val id = internalMessage.id();
             try {
                 mailbox.getActor().processWithObserver(internalMessage);
             }

memq-actor/src/main/java/io/appform/memq/actor/InternalMessage.java

@@ -1,15 +1,11 @@
 package io.appform.memq.actor;
 
-import lombok.Value;
-
 import java.util.Map;
 
-
-@Value
-class InternalMessage<M extends Message> {
-    String id;
-    long validTill;
-    long publishedAt;
-    Map<String,Object> headers;
-    M message;
+record InternalMessage<M extends Message>(
+        String id,
+        long validTill,
+        long publishedAt,
+        Map<String, Object> headers,
+        M message) {
 }

memq-actor/src/main/java/io/appform/memq/actor/Mailbox.java

@@ -88,7 +88,7 @@ public final boolean publish(final M message) {
             }
             val internalMessage = new InternalMessage<>(message.id(), message.validTill(),
                     System.currentTimeMillis(), message.headers(), message);
-            messages.putIfAbsent(internalMessage.getId(), internalMessage);
+            messages.putIfAbsent(internalMessage.id(), internalMessage);
             actor.getMessageDispatcher().triggerDispatch(this);
             return true;
         } finally {

memq-actor/src/main/java/io/appform/memq/actor/MessageMeta.java

@@ -9,7 +9,7 @@
 @Getter
 public class MessageMeta {
 
-    private AtomicInteger deliveryAttempt;
+    private final AtomicInteger deliveryAttempt;
     private final long publishedAt;
     private final long validTill;
     private final Map<String, Object> headers;

memq-actor/src/main/java/io/appform/memq/observer/ObserverMessageMeta.java

@@ -1,10 +1,4 @@
 package io.appform.memq.observer;
 
-import lombok.Value;
-
-@Value
-public class ObserverMessageMeta {
-    String id;
-    long validTill;
-    long publishedAt;
+public record ObserverMessageMeta(String id, long validTill, long publishedAt) {
 }

memq-actor/src/main/java/io/appform/memq/retry/RetryStrategy.java

@@ -8,7 +8,7 @@
 
 public abstract class RetryStrategy {
 
-    private RetryPolicy<Boolean> policy;
+    private final RetryPolicy<Boolean> policy;
 
     protected RetryStrategy(RetryPolicy<Boolean> policy) {
         this.policy = policy;

memq-actor/src/main/java/io/appform/memq/retry/RetryStrategyFactory.java

@@ -8,20 +8,14 @@
  */
 public class RetryStrategyFactory {
     public RetryStrategy create(RetryConfig config) {
-        return switch (config.getType()) {
-            case NO_RETRY -> new NoRetryStrategy((NoRetryConfig) config);
-            case TIME_LIMITED_EXPONENTIAL_BACKOFF -> new TimeLimitedExponentialWaitRetryStrategy(
-                    (TimeLimitedExponentialWaitRetryConfig) config);
-            case TIME_LIMITED_RANDOM_WAIT ->
-                    new TimeLimitedRandomWaitRetryStrategy((TimeLimitedRandomWaitRetryConfig) config);
-            case TIME_LIMITED_FIXED_WAIT ->
-                    new TimeLimitedFixedWaitRetryStrategy((TimeLimitedFixedWaitRetryConfig) config);
-            case COUNT_LIMITED_EXPONENTIAL_BACKOFF -> new CountLimitedExponentialWaitRetryStrategy(
-                    (CountLimitedExponentialWaitRetryConfig) config);
-            case COUNT_LIMITED_RANDOM_WAIT ->
-                    new CountLimitedRandomWaitRetryStrategy((CountLimitedRandomWaitRetryConfig) config);
-            case COUNT_LIMITED_FIXED_WAIT ->
-                    new CountLimitedFixedWaitRetryStrategy((CountLimitedFixedWaitRetryConfig) config);
+        return switch (config) {
+            case NoRetryConfig c -> new NoRetryStrategy(c);
+            case TimeLimitedExponentialWaitRetryConfig c -> new TimeLimitedExponentialWaitRetryStrategy(c);
+            case TimeLimitedRandomWaitRetryConfig c -> new TimeLimitedRandomWaitRetryStrategy(c);
+            case TimeLimitedFixedWaitRetryConfig c -> new TimeLimitedFixedWaitRetryStrategy(c);
+            case CountLimitedExponentialWaitRetryConfig c -> new CountLimitedExponentialWaitRetryStrategy(c);
+            case CountLimitedRandomWaitRetryConfig c -> new CountLimitedRandomWaitRetryStrategy(c);
+            case CountLimitedFixedWaitRetryConfig c -> new CountLimitedFixedWaitRetryStrategy(c);
         };
     }
 }

memq-actor/src/main/java/io/appform/memq/retry/config/CountLimitedExponentialWaitRetryConfig.java

@@ -1,11 +1,11 @@
 package io.appform.memq.retry.config;
 
 import io.appform.memq.retry.RetryType;
+import jakarta.validation.constraints.DecimalMin;
+import jakarta.validation.constraints.Min;
 import lombok.*;
 import lombok.extern.jackson.Jacksonized;
 
-import javax.validation.constraints.DecimalMin;
-import javax.validation.constraints.Min;
 import java.util.Set;
 
 @Value

memq-actor/src/main/java/io/appform/memq/retry/config/CountLimitedFixedWaitRetryConfig.java

@@ -1,10 +1,10 @@
 package io.appform.memq.retry.config;
 
 import io.appform.memq.retry.RetryType;
+import jakarta.validation.constraints.Min;
 import lombok.*;
 import lombok.extern.jackson.Jacksonized;
 
-import javax.validation.constraints.Min;
 import java.util.Set;
 
 @Value