Apply Palantir Java Format

Author: github-actions[bot]

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java

@@ -220,8 +220,7 @@ protected Object deleteAllMessages(QueueDetail queueDetail) {
             MessageDeleteRequest.builder().queueDetail(queueDetail).build());
   }
 
-  protected void registerQueueInternal(String queueName, QueueType type,
-      String... priorities) {
+  protected void registerQueueInternal(String queueName, QueueType type, String... priorities) {
     validateQueue(queueName);
     notNull(priorities, "priorities cannot be null");
     Map<String, Integer> priorityMap = new HashMap<>();

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.java

@@ -153,9 +153,8 @@ public void ensureStream(String streamName, List<String> subjects, QueueType que
       }
       try {
         StreamInfo existing = safeGetStreamInfo(streamName);
-        RetentionPolicy desired = queueType == QueueType.STREAM
-            ? RetentionPolicy.Limits
-            : RetentionPolicy.WorkQueue;
+        RetentionPolicy desired =
+            queueType == QueueType.STREAM ? RetentionPolicy.Limits : RetentionPolicy.WorkQueue;
         if (existing == null) {
           if (!config.isAutoCreateStreams()) {
             throw new RqueueNatsException(

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/js/NatsStreamValidator.java

@@ -163,8 +163,7 @@ private void tryEnsureConsumer(
     }
   }
 
-  private int tryEnsure(
-      List<String> failures, String streamName, String subject, QueueDetail q) {
+  private int tryEnsure(List<String> failures, String streamName, String subject, QueueDetail q) {
     try {
       provisioner.ensureStream(streamName, List.of(subject), q.getType());
       return 1;

rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/JetStreamMessageBrokerProducerOnlyIT.java

@@ -33,7 +33,8 @@ void enqueue_messagesAccumulateInStream() throws Exception {
         JetStreamMessageBroker.builder().connection(connection).build()) {
       int count = 10;
       for (int i = 0; i < count; i++) {
-        broker.enqueue(q, RqueueMessage.builder().id("m-" + i).message("payload-" + i).build());
+        broker.enqueue(
+            q, RqueueMessage.builder().id("m-" + i).message("payload-" + i).build());
       }
       assertEquals(count, broker.size(q), "all enqueued messages should be visible in the stream");
     }
@@ -76,13 +77,18 @@ void enqueueReactive_messagesAccumulateInStream() {
     try (JetStreamMessageBroker broker =
         JetStreamMessageBroker.builder().connection(connection).build()) {
       int count = 8;
-      Flux<Void> publishes = Flux.range(0, count).flatMap(i ->
-          broker.enqueueReactive(
-              q, RqueueMessage.builder().id("rm-" + i).message("reactive-payload-" + i).build()));
+      Flux<Void> publishes = Flux.range(0, count)
+          .flatMap(i -> broker.enqueueReactive(
+              q,
+              RqueueMessage.builder()
+                  .id("rm-" + i)
+                  .message("reactive-payload-" + i)
+                  .build()));
 
       StepVerifier.create(publishes).verifyComplete();
 
-      assertEquals(count, broker.size(q), "all reactively enqueued messages should be in the stream");
+      assertEquals(
+          count, broker.size(q), "all reactively enqueued messages should be in the stream");
     }
   }
 
@@ -97,7 +103,8 @@ void mixedEnqueue_allVariantsLandInCorrectStreams() {
 
       // 3 plain messages on the main queue
       for (int i = 0; i < 3; i++) {
-        broker.enqueue(mainQ, RqueueMessage.builder().id("plain-" + i).message("p" + i).build());
+        broker.enqueue(
+            mainQ, RqueueMessage.builder().id("plain-" + i).message("p" + i).build());
       }
       // 2 priority messages on the "high" sub-queue
       for (int i = 0; i < 2; i++) {
@@ -107,9 +114,8 @@ void mixedEnqueue_allVariantsLandInCorrectStreams() {
             RqueueMessage.builder().id("high-" + i).message("h" + i).build());
       }
       // 1 reactive message on the main queue
-      StepVerifier.create(
-              broker.enqueueReactive(
-                  mainQ, RqueueMessage.builder().id("react-0").message("r0").build()))
+      StepVerifier.create(broker.enqueueReactive(
+              mainQ, RqueueMessage.builder().id("react-0").message("r0").build()))
           .verifyComplete();
 
       assertEquals(4L, broker.size(mainQ), "main stream: 3 plain + 1 reactive");

rqueue-nats/src/test/java/com/github/sonus21/rqueue/nats/JetStreamQueueModeIT.java

@@ -55,35 +55,27 @@ void queueMode_queue_createsWorkQueueStream() throws Exception {
     String streamName = "rqueue-" + "qm-queue-" + System.nanoTime();
     String subject = "rqueue." + "qm-queue-" + System.nanoTime();
     JetStreamManagement jsm = connection.jetStreamManagement();
-    NatsProvisioner provisioner =
-        new NatsProvisioner(connection, jsm, RqueueNatsConfig.defaults());
+    NatsProvisioner provisioner = new NatsProvisioner(connection, jsm, RqueueNatsConfig.defaults());
 
     provisioner.ensureStream(streamName, List.of(subject), QueueType.QUEUE);
 
-    RetentionPolicy actual =
-        jsm.getStreamInfo(streamName).getConfiguration().getRetentionPolicy();
+    RetentionPolicy actual = jsm.getStreamInfo(streamName).getConfiguration().getRetentionPolicy();
     assertEquals(
-        RetentionPolicy.WorkQueue,
-        actual,
-        "QUEUE mode must create a WorkQueue-retention stream");
+        RetentionPolicy.WorkQueue, actual, "QUEUE mode must create a WorkQueue-retention stream");
   }
 
   @Test
   void queueMode_stream_createsLimitsStream() throws Exception {
     String streamName = "rqueue-" + "qm-stream-" + System.nanoTime();
     String subject = "rqueue." + "qm-stream-" + System.nanoTime();
     JetStreamManagement jsm = connection.jetStreamManagement();
-    NatsProvisioner provisioner =
-        new NatsProvisioner(connection, jsm, RqueueNatsConfig.defaults());
+    NatsProvisioner provisioner = new NatsProvisioner(connection, jsm, RqueueNatsConfig.defaults());
 
     provisioner.ensureStream(streamName, List.of(subject), QueueType.STREAM);
 
-    RetentionPolicy actual =
-        jsm.getStreamInfo(streamName).getConfiguration().getRetentionPolicy();
+    RetentionPolicy actual = jsm.getStreamInfo(streamName).getConfiguration().getRetentionPolicy();
     assertEquals(
-        RetentionPolicy.Limits,
-        actual,
-        "STREAM mode must create a Limits-retention stream");
+        RetentionPolicy.Limits, actual, "STREAM mode must create a Limits-retention stream");
   }
 
   // ---- Contract 2: consumer reuse preserves delivery position -----------
@@ -145,7 +137,8 @@ void queueMode_consumerReuse_preservesDeliveryPosition() throws Exception {
           cd.getMaxAckPending());
 
       // Verify the consumer info still reflects the already-delivered messages.
-      ConsumerInfo info = jsm.getConsumerInfo(RqueueNatsConfig.defaults().getStreamPrefix() + q.getName(), consumerName);
+      ConsumerInfo info = jsm.getConsumerInfo(
+          RqueueNatsConfig.defaults().getStreamPrefix() + q.getName(), consumerName);
       long numAcked = info.getNumAckPending() == 0
           ? total - info.getNumPending()
           : total - info.getNumPending() - info.getNumAckPending();
@@ -154,8 +147,8 @@ void queueMode_consumerReuse_preservesDeliveryPosition() throws Exception {
       assertEquals(
           total - firstBatch,
           remaining,
-          "consumer position must be preserved across ensureConsumer calls; "
-              + "remaining=" + remaining + " but expected " + (total - firstBatch));
+          "consumer position must be preserved across ensureConsumer calls; " + "remaining="
+              + remaining + " but expected " + (total - firstBatch));
     }
   }
 
@@ -179,8 +172,7 @@ void queueMode_queue_competingConsumers_eachMessageDeliveredOnce() throws Except
       for (int t = 0; t < 2; t++) {
         pool.submit(() -> {
           for (int round = 0; round < 100 && done.getCount() > 0; round++) {
-            List<RqueueMessage> msgs =
-                broker.pop(q, sharedConsumer, 5, Duration.ofMillis(300));
+            List<RqueueMessage> msgs = broker.pop(q, sharedConsumer, 5, Duration.ofMillis(300));
             for (RqueueMessage m : msgs) {
               if (seen.add(m.getId())) {
                 done.countDown();
@@ -194,9 +186,7 @@ void queueMode_queue_competingConsumers_eachMessageDeliveredOnce() throws Except
       pool.shutdownNow();
 
       assertEquals(
-          total,
-          seen.size(),
-          "QUEUE mode: each message must be delivered to exactly one worker");
+          total, seen.size(), "QUEUE mode: each message must be delivered to exactly one worker");
     }
   }
 
@@ -219,9 +209,7 @@ void queueMode_stream_fanOut_everyConsumerReceivesAllMessages() throws Exception
       Set<String> listenerTwoSeen = drain(broker, q, "listener-svc-2", total);
 
       assertEquals(
-          total,
-          listenerOneSeen.size(),
-          "STREAM mode: listener-svc-1 must receive all messages");
+          total, listenerOneSeen.size(), "STREAM mode: listener-svc-1 must receive all messages");
       assertEquals(
           total,
           listenerTwoSeen.size(),