IPC dead letter queue can now be deactivated

Author: jlangch

src/main/java/com/github/jlangch/venice/impl/functions/IPCFunctions.java

@@ -133,10 +133,11 @@ public class IPCFunctions {
                                                         " Defaults to `false`.|\n" +
                         "| :write-ahead-log-compact b   | If `true` compacts the write-ahead-logs at server start.¶" +
                                                         " Defaults to `false`.|\n" +
-                        "| :authenticator a             | An authenticator. If an authenticator is used encryption must " +
+                        "| :authenticator a             | An authenticator. If an authenticator is used encryption must" +
                                                         " be enabled to safely transmit users credentials!¶" +
                                                         " Defaults to `nil`.|\n" +
-                        "| :dead-letter-queue-size n    | The dead letter queue size.¶Defaults to 100|\n" +
+                        "| :dead-letter-queue-size n    | The dead letter queue size. A size of 0 deactivates the dead" +
+                                                        " letter queue. ¶Defaults to 100|\n" +
                         "| :socket-snd-buf-size n       | The server socket's send buffer size.¶" +
                                                         " Defaults to `-1` (use the sockets default buf size).¶" +
                                                         " The size can be specified as a number like `64536`" +

src/main/java/com/github/jlangch/venice/util/ipc/ServerConfig.java

@@ -360,11 +360,12 @@ public Builder maxFunctions(final int maxFunctions) {
          *
          * <p>Defaults to 100
          *
-         * @param size the dead letter queue's size.
+         * @param size the dead letter queue's size. A size of 0 deactivates
+         *             the dead letter queue
          * @return this builder
          */
         public Builder deadLetterQueueSize(final int size) {
-            this.deadLetterQueueSize = Math.min(10_000, Math.max(10, size));
+            this.deadLetterQueueSize = Math.min(10_000, Math.max(0, size));
             return this;
         }
 

src/main/java/com/github/jlangch/venice/util/ipc/impl/ServerQueueManager.java

@@ -22,6 +22,7 @@
 package com.github.jlangch.venice.util.ipc.impl;
 
 import static com.github.jlangch.venice.util.ipc.QueuePersistence.DURABLE;
+import static com.github.jlangch.venice.util.ipc.QueueType.CIRCULAR;
 
 import java.util.HashMap;
 import java.util.List;
@@ -38,6 +39,7 @@
 import com.github.jlangch.venice.util.ipc.ServerConfig;
 import com.github.jlangch.venice.util.ipc.impl.dest.queue.CircularBuffer;
 import com.github.jlangch.venice.util.ipc.impl.dest.queue.IpcQueue;
+import com.github.jlangch.venice.util.ipc.impl.dest.queue.NullQueue;
 import com.github.jlangch.venice.util.ipc.impl.util.ServerLogger;
 import com.github.jlangch.venice.util.ipc.impl.wal.WalQueueManager;
 
@@ -299,13 +301,18 @@ public void close() {
     }
 
     private IpcQueue<Message> creatDeadLetterQueue(final int size) {
-        final IpcQueue<Message> q = new CircularBuffer<Message>(
+        final IpcQueue<Message> q = size > 0
+                                        ? new CircularBuffer<Message>(
                                             DEAD_LETTER_QUEUE_NAME,
                                             size,
-                                            false);
+                                            false)
+                                        : new NullQueue<Message>(
+                                            DEAD_LETTER_QUEUE_NAME,
+                                            CIRCULAR);
         q.updateAcls(
                 new HashMap<String,Acl>(),
                 new Acl(DEAD_LETTER_QUEUE_NAME, null, AccessMode.DENY)); // admin only
+
         return q;
     }
 

src/main/java/com/github/jlangch/venice/util/ipc/impl/dest/queue/NullQueue.java

@@ -0,0 +1,94 @@
+/*   __    __         _
+ *   \ \  / /__ _ __ (_) ___ ___
+ *    \ \/ / _ \ '_ \| |/ __/ _ \
+ *     \  /  __/ | | | | (_|  __/
+ *      \/ \___|_| |_|_|\___\___|
+ *
+ *
+ * Copyright 2017-2026 Venice
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.github.jlangch.venice.util.ipc.impl.dest.queue;
+
+import java.util.concurrent.TimeUnit;
+
+import com.github.jlangch.venice.util.ipc.QueueType;
+import com.github.jlangch.venice.util.ipc.impl.Destination;
+
+
+public class NullQueue<T> extends Destination implements IpcQueue<T> {
+
+    public NullQueue(final String name, final QueueType type) {
+        super(name);
+        this.type = type;
+    }
+
+
+    @Override
+    public QueueType type() {
+        return type;
+    }
+
+    @Override
+    public boolean isTemporary() {
+        return true;
+    }
+
+    @Override
+    public boolean isDurable() {
+        return false;
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return true;
+    }
+
+    @Override
+    public int size() {
+        return 0;
+    }
+
+    @Override
+    public int capacity() {
+        return 0;
+    }
+
+    @Override
+    public T poll() throws InterruptedException {
+        return null;
+    }
+
+    @Override
+    public T poll(long timeout, TimeUnit unit) throws InterruptedException {
+        return null;
+    }
+
+    @Override
+    public boolean offer(T item) throws InterruptedException {
+        return false;
+    }
+
+    @Override
+    public boolean offer(T item, long timeout, TimeUnit unit) throws InterruptedException {
+        return false;
+    }
+
+    @Override
+    public void onRemove() {
+    }
+
+
+    private final QueueType type;
+}