This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new ba28701069f7 CAMEL-25346: camel-console - a dev console capacity set
from configuration breaks the trace, receive, event and sql-trace consoles
(#27380)
ba28701069f7 is described below
commit ba28701069f75b71cdd885440af8e3cc284fe8ff
Author: Federico Mariani <[email protected]>
AuthorDate: Mon Oct 5 20:37:59 2026 +0200
CAMEL-25346: camel-console - a dev console capacity set from configuration
breaks the trace, receive, event and sql-trace consoles (#27380)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/impl/console/ConsoleHelper.java | 69 +++++++++
.../apache/camel/impl/console/EventConsole.java | 54 +++----
.../camel/impl/console/ReceiveDevConsole.java | 32 ++--
.../camel/impl/console/SqlTraceDevConsole.java | 20 ++-
.../apache/camel/impl/console/TraceDevConsole.java | 34 ++---
.../camel/impl/console/ConsoleHelperTest.java | 37 +++++
.../camel/impl/console/DevConsoleCapacityTest.java | 170 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 7 +
8 files changed, 355 insertions(+), 68 deletions(-)
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/ConsoleHelper.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/ConsoleHelper.java
index bb7f1beaad10..0720e5ff00f7 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/ConsoleHelper.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/ConsoleHelper.java
@@ -19,8 +19,12 @@ package org.apache.camel.impl.console;
import java.io.LineNumberReader;
import java.io.Reader;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.CamelContext;
import org.apache.camel.spi.Resource;
@@ -147,4 +151,69 @@ public final class ConsoleHelper {
return location;
}
+ /**
+ * Validates the capacity of a console buffer.
+ *
+ * @throws IllegalArgumentException if the capacity is not between the
given minimum and 1000
+ */
+ static void checkCapacity(int capacity, int min) {
+ if (capacity > 1000 || capacity < min) {
+ throw new IllegalArgumentException("Capacity must be between " +
min + " and 1000");
+ }
+ }
+
+ /**
+ * Adds the element to the bounded queue, removing the oldest elements to
make room.
+ */
+ static <T> void offerLast(Queue<T> queue, T element) {
+ while (!queue.offer(element)) {
+ queue.poll();
+ }
+ }
+
+ /**
+ * Creates a bounded queue of the given capacity holding the newest
elements of the given queue.
+ */
+ static <T> Queue<T> resize(Queue<T> queue, int capacity) {
+ Queue<T> answer = new LinkedBlockingQueue<>(Math.max(1, capacity));
+ for (T element : queue) {
+ offerLast(answer, element);
+ }
+ return answer;
+ }
+
+ /**
+ * Creates a ring buffer of the given capacity holding the newest entries
of the given ring buffer, so that slot 0
+ * is the next one to write.
+ *
+ * @param ring the ring buffer
+ * @param next the next slot to write in the given ring buffer
+ */
+ static <T> T[] resize(T[] ring, int next, int capacity) {
+ int length = Math.max(1, capacity);
+ List<T> entries = new ArrayList<>(ring.length);
+ for (int i = 0; i < ring.length; i++) {
+ T entry = ring[(next + i) % ring.length];
+ if (entry != null) {
+ entries.add(entry);
+ }
+ }
+ T[] answer = Arrays.copyOf(ring, length);
+ Arrays.fill(answer, null);
+ // the newest entries, oldest first, at the end: slot 0 holds the
oldest (or nothing) and is written next
+ int kept = Math.min(entries.size(), length);
+ for (int i = 0; i < kept; i++) {
+ answer[length - kept + i] = entries.get(entries.size() - kept + i);
+ }
+ return answer;
+ }
+
+ /**
+ * Returns the slot to write in a ring buffer of the given length, and
moves the position to the next slot.
+ */
+ static int nextSlot(AtomicInteger pos, int length) {
+ // the position may come from a larger ring buffer that has just been
resized
+ return pos.getAndUpdate(operand -> (operand + 1) % length) % length;
+ }
+
}
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/EventConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/EventConsole.java
index af8a56eec145..a41590aac968 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/EventConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/EventConsole.java
@@ -53,11 +53,11 @@ public class EventConsole extends AbstractDevConsole {
description = "Maximum capacity of last number of events to
capture (capacity must be between 25 and 1000)")
private int capacity = 25;
- private CamelEvent[] events;
+ private volatile CamelEvent[] events;
private final AtomicInteger posEvents = new AtomicInteger();
- private CamelEvent.RouteEvent[] routeEvents;
+ private volatile CamelEvent.RouteEvent[] routeEvents;
private final AtomicInteger posRoutes = new AtomicInteger();
- private CamelEvent.ExchangeEvent[] exchangeEvents;
+ private volatile CamelEvent.ExchangeEvent[] exchangeEvents;
private final AtomicInteger posExchanges = new AtomicInteger();
private final ConsoleEventNotifier listener = new ConsoleEventNotifier();
@@ -70,14 +70,20 @@ public class EventConsole extends AbstractDevConsole {
}
public void setCapacity(int capacity) {
+ if (events != null) {
+ // Camel Main configures the console after the registry has
started it: resize its buffers (an entry
+ // added meanwhile may be lost, which only matters while
configuring)
+ ConsoleHelper.checkCapacity(capacity, 25);
+ this.events = ConsoleHelper.resize(events, posEvents.getAndSet(0),
capacity);
+ this.routeEvents = ConsoleHelper.resize(routeEvents,
posRoutes.getAndSet(0), capacity);
+ this.exchangeEvents = ConsoleHelper.resize(exchangeEvents,
posExchanges.getAndSet(0), capacity);
+ }
this.capacity = capacity;
}
@Override
protected void doInit() throws Exception {
- if (capacity > 1000 || capacity < 25) {
- throw new IllegalArgumentException("Capacity must be between 25
and 1000");
- }
+ ConsoleHelper.checkCapacity(capacity, 25);
this.events = new CamelEvent[capacity];
this.routeEvents = new CamelEvent.RouteEvent[capacity];
this.exchangeEvents = new CamelEvent.ExchangeEvent[capacity];
@@ -96,37 +102,32 @@ public class EventConsole extends AbstractDevConsole {
protected String doCallText(Map<String, Object> options) {
StringBuilder sb = new StringBuilder();
- int pos = posEvents.get();
- sb.append(appendTextEvents(events, "Camel", pos, capacity));
+ sb.append(appendTextEvents(events, "Camel", posEvents.get()));
sb.append("\n");
- pos = posRoutes.get();
- sb.append(appendTextEvents(routeEvents, "Route", pos, capacity));
+ sb.append(appendTextEvents(routeEvents, "Route", posRoutes.get()));
sb.append("\n");
- pos = posExchanges.get();
- sb.append(appendTextEvents(exchangeEvents, "Exchange", pos, capacity));
+ sb.append(appendTextEvents(exchangeEvents, "Exchange",
posExchanges.get()));
sb.append("\n");
return sb.toString();
}
protected Map<String, Object> doCallJson(Map<String, Object> options) {
- int pos = posEvents.get();
- List<EventEntry> arr = appendJSonEvents(events, pos, capacity);
+ List<EventEntry> arr = appendJSonEvents(events, posEvents.get());
List<EventEntry> eventsOut = !arr.isEmpty() ? arr : null;
- pos = posRoutes.get();
- arr = appendJSonEvents(routeEvents, pos, capacity);
+ arr = appendJSonEvents(routeEvents, posRoutes.get());
List<EventEntry> routeEventsOut = !arr.isEmpty() ? arr : null;
- pos = posExchanges.get();
- arr = appendJSonEvents(exchangeEvents, pos, capacity);
+ arr = appendJSonEvents(exchangeEvents, posExchanges.get());
List<EventEntry> exchangeEventsOut = !arr.isEmpty() ? arr : null;
Response response = new Response(eventsOut, routeEventsOut,
exchangeEventsOut);
return JsonRecordSupport.toJsonObject(response);
}
- private static String appendTextEvents(CamelEvent[] events, String kind,
int cursor, int capacity) {
+ private static String appendTextEvents(CamelEvent[] events, String kind,
int cursor) {
+ int capacity = events.length;
StringBuilder sb = new StringBuilder();
int pos = 0;
int added = 0;
@@ -153,7 +154,8 @@ public class EventConsole extends AbstractDevConsole {
return sb.toString();
}
- private static List<EventEntry> appendJSonEvents(CamelEvent[] events, int
cursor, int capacity) {
+ private static List<EventEntry> appendJSonEvents(CamelEvent[] events, int
cursor) {
+ int capacity = events.length;
List<EventEntry> arr = new ArrayList<>();
int pos = 0;
// cursor is at last event, so move to back
@@ -191,15 +193,15 @@ public class EventConsole extends AbstractDevConsole {
public void notify(CamelEvent event) throws Exception {
if (event instanceof CamelEvent.ExchangeEvent) {
CamelEvent.ExchangeEvent ce = (CamelEvent.ExchangeEvent) event;
- int pos = posExchanges.getAndUpdate(operand -> ++operand %
capacity);
- exchangeEvents[pos] = ce;
+ CamelEvent.ExchangeEvent[] ring = exchangeEvents;
+ ring[ConsoleHelper.nextSlot(posExchanges, ring.length)] = ce;
} else if (event instanceof CamelEvent.RouteEvent) {
CamelEvent.RouteEvent re = (CamelEvent.RouteEvent) event;
- int pos = posRoutes.getAndUpdate(operand -> ++operand %
capacity);
- routeEvents[pos] = re;
+ CamelEvent.RouteEvent[] ring = routeEvents;
+ ring[ConsoleHelper.nextSlot(posRoutes, ring.length)] = re;
} else {
- int pos = posEvents.getAndUpdate(operand -> ++operand %
capacity);
- events[pos] = event;
+ CamelEvent[] ring = events;
+ ring[ConsoleHelper.nextSlot(posEvents, ring.length)] = event;
}
}
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java
index 70654e151b76..0c38c24a20ed 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/ReceiveDevConsole.java
@@ -96,7 +96,7 @@ public class ReceiveDevConsole extends AbstractDevConsole {
private final List<Consumer> consumers = new ArrayList<>();
private final AtomicBoolean enabled = new AtomicBoolean();
private final AtomicLong uuid = new AtomicLong();
- private Queue<JsonObject> queue;
+ private volatile Queue<JsonObject> queue;
private long firstTimestamp;
private long lastTimestamp;
@@ -109,6 +109,13 @@ public class ReceiveDevConsole extends AbstractDevConsole {
}
public void setCapacity(int capacity) {
+ Queue<JsonObject> q = queue;
+ if (q != null) {
+ // Camel Main configures the console after the registry has
started it: resize its buffers (an entry
+ // added meanwhile may be lost, which only matters while
configuring)
+ ConsoleHelper.checkCapacity(capacity, 50);
+ this.queue = ConsoleHelper.resize(q, capacity);
+ }
this.capacity = capacity;
}
@@ -130,9 +137,7 @@ public class ReceiveDevConsole extends AbstractDevConsole {
@Override
protected void doInit() throws Exception {
- if (capacity > 1000 || capacity < 50) {
- throw new IllegalArgumentException("Capacity must be between 50
and 1000");
- }
+ ConsoleHelper.checkCapacity(capacity, 50);
this.queue = new LinkedBlockingQueue<>(capacity);
}
@@ -153,10 +158,11 @@ public class ReceiveDevConsole extends AbstractDevConsole
{
String dump = optionString(options, DUMP);
if ("true".equals(dump)) {
+ Queue<JsonObject> q = queue;
JsonArray arr = new JsonArray();
- arr.addAll(queue);
+ arr.addAll(q);
if (removeOnDump) {
- queue.clear();
+ q.clear();
}
JsonObject jo = (JsonObject) arr.get(0);
firstTimestamp = jo.getLongOrDefault("timestamp", 0);
@@ -205,9 +211,10 @@ public class ReceiveDevConsole extends AbstractDevConsole {
protected Map<String, Object> doCallJson(Map<String, Object> options) {
String dump = optionString(options, DUMP);
if ("true".equals(dump)) {
- List<Map<String, Object>> messages = new ArrayList<>(queue);
+ Queue<JsonObject> q = queue;
+ List<Map<String, Object>> messages = new ArrayList<>(q);
if (removeOnDump) {
- queue.clear();
+ q.clear();
}
JsonObject first = (JsonObject) messages.get(0);
firstTimestamp = first.getLongOrDefault("timestamp", 0);
@@ -282,14 +289,7 @@ public class ReceiveDevConsole extends AbstractDevConsole {
lastTimestamp = exchange.getMessage().getMessageTimestamp();
json.put("timestamp", lastTimestamp);
- // ensure there is space on the queue by polling until at least single
slot is free
- int drain = queue.size() - capacity + 1;
- if (drain > 0) {
- for (int i = 0; i < drain; i++) {
- queue.poll();
- }
- }
- queue.add(json);
+ ConsoleHelper.offerLast(queue, json);
}
protected static Endpoint findMatchingEndpoint(CamelContext camelContext,
String endpoint) {
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/SqlTraceDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/SqlTraceDevConsole.java
index cc6b9afcb86e..076cd3728a80 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/SqlTraceDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/SqlTraceDevConsole.java
@@ -49,7 +49,7 @@ public class SqlTraceDevConsole extends AbstractDevConsole {
description = "Maximum capacity of traced SQL statements
(capacity must be between 25 and 1000)")
private int capacity = 200;
- private StatementEntry[] events;
+ private volatile StatementEntry[] events;
private final AtomicInteger pos = new AtomicInteger();
private final ConsoleEventNotifier listener = new ConsoleEventNotifier();
@@ -94,14 +94,18 @@ public class SqlTraceDevConsole extends AbstractDevConsole {
}
public void setCapacity(int capacity) {
+ if (events != null) {
+ // Camel Main configures the console after the registry has
started it: resize its buffers (an entry
+ // added meanwhile may be lost, which only matters while
configuring)
+ ConsoleHelper.checkCapacity(capacity, 25);
+ this.events = ConsoleHelper.resize(events, pos.getAndSet(0),
capacity);
+ }
this.capacity = capacity;
}
@Override
protected void doInit() throws Exception {
- if (capacity > 1000 || capacity < 25) {
- throw new IllegalArgumentException("Capacity must be between 25
and 1000");
- }
+ ConsoleHelper.checkCapacity(capacity, 25);
this.events = new StatementEntry[capacity];
}
@@ -191,11 +195,13 @@ public class SqlTraceDevConsole extends
AbstractDevConsole {
private List<StatementEntry> collectEvents() {
List<StatementEntry> list = new ArrayList<>();
+ StatementEntry[] ring = events;
+ int capacity = ring.length;
int cursor = pos.get();
// cursor points to the NEXT write slot, so walk backward from cursor-1
for (int i = 0; i < capacity; i++) {
cursor = (cursor - 1 + capacity) % capacity;
- StatementEntry event = events[cursor];
+ StatementEntry event = ring[cursor];
if (event != null) {
list.add(event);
}
@@ -345,8 +351,8 @@ public class SqlTraceDevConsole extends AbstractDevConsole {
event.getTimestamp(), exchange.getExchangeId(),
exchange.getFromRouteId(), nodeId, location,
uri, finalQuery, category, ese.getTimeTaken(),
exchange.isFailed(), rowCount, updateCount);
- int p = pos.getAndUpdate(operand -> ++operand % capacity);
- events[p] = entry;
+ StatementEntry[] ring = events;
+ ring[ConsoleHelper.nextSlot(pos, ring.length)] = entry;
}
}
}
diff --git
a/core/camel-console/src/main/java/org/apache/camel/impl/console/TraceDevConsole.java
b/core/camel-console/src/main/java/org/apache/camel/impl/console/TraceDevConsole.java
index a8fdf375fe34..bd36449926f9 100644
---
a/core/camel-console/src/main/java/org/apache/camel/impl/console/TraceDevConsole.java
+++
b/core/camel-console/src/main/java/org/apache/camel/impl/console/TraceDevConsole.java
@@ -67,7 +67,7 @@ public class TraceDevConsole extends AbstractDevConsole {
javaType = "java.lang.String", enums = "true,false")
public static final String DUMP = "dump";
- private Queue<BacklogTracerEventMessage> queue;
+ private volatile Queue<BacklogTracerEventMessage> queue;
public TraceDevConsole() {
super("camel", "trace", "Camel Tracing", "Trace routed messages");
@@ -78,14 +78,19 @@ public class TraceDevConsole extends AbstractDevConsole {
}
public void setCapacity(int capacity) {
+ Queue<BacklogTracerEventMessage> q = queue;
+ if (q != null) {
+ // Camel Main configures the console after the registry has
started it: resize its buffers (an entry
+ // added meanwhile may be lost, which only matters while
configuring)
+ ConsoleHelper.checkCapacity(capacity, 50);
+ this.queue = ConsoleHelper.resize(q, capacity);
+ }
this.capacity = capacity;
}
@Override
protected void doInit() throws Exception {
- if (capacity > 1000 || capacity < 50) {
- throw new IllegalArgumentException("Capacity must be between 50
and 1000");
- }
+ ConsoleHelper.checkCapacity(capacity, 50);
this.queue = new LinkedBlockingQueue<>(capacity);
}
@@ -97,10 +102,11 @@ public class TraceDevConsole extends AbstractDevConsole {
BacklogTracer tracer =
getCamelContext().getCamelContextExtension().getContextPlugin(BacklogTracer.class);
if (tracer != null) {
if ("true".equalsIgnoreCase(dump)) {
+ Queue<BacklogTracerEventMessage> q = queue;
for (BacklogTracerEventMessage t :
tracer.dumpAllTracedMessages()) {
- addMessage(t);
+ ConsoleHelper.offerLast(q, t);
}
- for (BacklogTracerEventMessage t : queue) {
+ for (BacklogTracerEventMessage t : q) {
String json = t.toJSon(0);
sb.append(json).append("\n");
}
@@ -136,17 +142,6 @@ public class TraceDevConsole extends AbstractDevConsole {
return sb.toString();
}
- private void addMessage(BacklogTracerEventMessage message) {
- // ensure there is space on the queue by polling until at least single
slot is free
- int drain = queue.size() - capacity + 1;
- if (drain > 0) {
- for (int i = 0; i < drain; i++) {
- queue.poll();
- }
- }
- queue.add(message);
- }
-
protected Map<String, Object> doCallJson(Map<String, Object> options) {
String enabled = optionString(options, ENABLED);
String dump = optionString(options, DUMP);
@@ -157,11 +152,12 @@ public class TraceDevConsole extends AbstractDevConsole {
BacklogTracer tracer =
getCamelContext().getCamelContextExtension().getContextPlugin(BacklogTracer.class);
if (tracer != null) {
if ("true".equalsIgnoreCase(dump)) {
+ Queue<BacklogTracerEventMessage> q = queue;
for (BacklogTracerEventMessage t :
tracer.dumpAllTracedMessages()) {
- addMessage(t);
+ ConsoleHelper.offerLast(q, t);
}
List<Map<String, Object>> traces = new ArrayList<>();
- for (BacklogTracerEventMessage t : queue) {
+ for (BacklogTracerEventMessage t : q) {
traces.add((JsonObject) t.asJSon());
}
response = new Response(
diff --git
a/core/camel-console/src/test/java/org/apache/camel/impl/console/ConsoleHelperTest.java
b/core/camel-console/src/test/java/org/apache/camel/impl/console/ConsoleHelperTest.java
index 84893bdad92d..da0e676def8c 100644
---
a/core/camel-console/src/test/java/org/apache/camel/impl/console/ConsoleHelperTest.java
+++
b/core/camel-console/src/test/java/org/apache/camel/impl/console/ConsoleHelperTest.java
@@ -18,6 +18,10 @@ package org.apache.camel.impl.console;
import java.io.Reader;
import java.io.StringReader;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.camel.util.json.JsonArray;
import org.apache.camel.util.json.JsonObject;
@@ -117,4 +121,37 @@ public class ConsoleHelperTest {
JsonArray result = ConsoleHelper.loadSourceAsJson(nullReader, 1);
Assertions.assertNull(result);
}
+
+ @Test
+ public void testResizeRingKeepsNewestEntries() {
+ // 5 slots, written 7 times: next slot is 2, oldest to newest is 3..7
+ String[] ring = { "6", "7", "3", "4", "5" };
+
+ String[] larger = ConsoleHelper.resize(ring, 2, 8);
+ Assertions.assertArrayEquals(new String[] { null, null, null, "3",
"4", "5", "6", "7" }, larger);
+
+ String[] smaller = ConsoleHelper.resize(ring, 2, 3);
+ Assertions.assertArrayEquals(new String[] { "5", "6", "7" }, smaller);
+ }
+
+ @Test
+ public void testNextSlotAfterResize() {
+ AtomicInteger pos = new AtomicInteger(7);
+ // a position from a larger ring buffer stays within the smaller one
+ Assertions.assertEquals(1, ConsoleHelper.nextSlot(pos, 3));
+ Assertions.assertEquals(2, ConsoleHelper.nextSlot(pos, 3));
+ Assertions.assertEquals(0, ConsoleHelper.nextSlot(pos, 3));
+ }
+
+ @Test
+ public void testResizeQueueKeepsNewestElements() {
+ Queue<Integer> queue = new LinkedBlockingQueue<>(5);
+ for (int i = 1; i <= 7; i++) {
+ ConsoleHelper.offerLast(queue, i);
+ }
+ Assertions.assertEquals(List.of(3, 4, 5, 6, 7), List.copyOf(queue));
+
+ Assertions.assertEquals(List.of(5, 6, 7),
List.copyOf(ConsoleHelper.resize(queue, 3)));
+ Assertions.assertEquals(List.of(3, 4, 5, 6, 7),
List.copyOf(ConsoleHelper.resize(queue, 10)));
+ }
}
diff --git
a/core/camel-console/src/test/java/org/apache/camel/impl/console/DevConsoleCapacityTest.java
b/core/camel-console/src/test/java/org/apache/camel/impl/console/DevConsoleCapacityTest.java
new file mode 100644
index 000000000000..96b17fe094f8
--- /dev/null
+++
b/core/camel-console/src/test/java/org/apache/camel/impl/console/DevConsoleCapacityTest.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.camel.impl.console;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.IntConsumer;
+import java.util.function.IntSupplier;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.console.DevConsole;
+import org.apache.camel.console.DevConsoleRegistry;
+import org.apache.camel.util.json.JsonObject;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Camel Main sets {@code camel.devConsole.<id>.capacity} on a console the
registry has already started, so the console
+ * must resize the buffers it allocated with the default capacity.
+ */
+public class DevConsoleCapacityTest extends ContextTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.setBacklogTracing(true);
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").routeId("myRoute").log("${body}").to("mock:result");
+ }
+ };
+ }
+
+ private <T extends DevConsole> T resolve(String id) {
+ // the registry starts the console, as when Camel Main configures it
+ return (T)
context.getCamelContextExtension().getContextPlugin(DevConsoleRegistry.class).resolveById(id);
+ }
+
+ private void sendMessages(String uri, int count) {
+ for (int i = 0; i < count; i++) {
+ template.sendBody(uri, "Hello " + i);
+ }
+ }
+
+ @Test
+ public void testTraceCapacitySetAfterStart() {
+ TraceDevConsole console = resolve("trace");
+ console.setCapacity(1000);
+
+ // each dump takes up to 100 events from the tracer (its backlog
size): the second one used to fail with
+ // "Queue full" as the queue still had the default 100 slots
+ int traces = 0;
+ for (int round = 0; round < 3; round++) {
+ sendMessages("direct:start", 50);
+ JsonObject out = (JsonObject)
console.call(DevConsole.MediaType.JSON, Map.of("dump", "true"));
+ traces = out.getCollection("traces").size();
+ }
+ assertEquals(300, traces);
+ }
+
+ @Test
+ public void testTraceCapacityLoweredAfterStart() {
+ TraceDevConsole console = resolve("trace");
+ sendMessages("direct:start", 50);
+ console.call(DevConsole.MediaType.JSON, Map.of("dump", "true"));
+
+ console.setCapacity(50);
+ sendMessages("direct:start", 50);
+ JsonObject out = (JsonObject) console.call(DevConsole.MediaType.JSON,
Map.of("dump", "true"));
+ assertEquals(50, out.getCollection("traces").size());
+ }
+
+ @Test
+ public void testReceiveCapacitySetAfterStart() {
+ ReceiveDevConsole console = resolve("receive");
+ console.setCapacity(1000);
+ console.call(DevConsole.MediaType.JSON, Map.of("enabled", "true",
"endpoint", "direct:incoming"));
+
+ sendMessages("direct:incoming", 150);
+
+ JsonObject out = (JsonObject) console.call(DevConsole.MediaType.JSON,
Map.of("dump", "true"));
+ List<JsonObject> messages = out.getCollection("messages");
+ assertEquals(150, messages.size());
+ assertEquals(150, messages.get(149).getLong("uid"));
+ }
+
+ @Test
+ public void testEventCapacitySetAfterStart() {
+ EventConsole console = resolve("event");
+ console.setCapacity(1000);
+
+ // used to fail with ArrayIndexOutOfBoundsException once past the
default 25 slots
+ sendMessages("direct:start", 50);
+
+ JsonObject out = (JsonObject) console.call(DevConsole.MediaType.JSON);
+ assertTrue(out.getCollection("exchangeEvents").size() > 25);
+ assertTrue(((String)
console.call(DevConsole.MediaType.TEXT)).contains("Exchange Events"));
+ }
+
+ @Test
+ public void testEventCapacityLoweredAfterStart() {
+ EventConsole console = resolve("event");
+ console.setCapacity(100);
+ sendMessages("direct:start", 30);
+
+ console.setCapacity(25);
+ sendMessages("direct:start", 30);
+
+ JsonObject out = (JsonObject) console.call(DevConsole.MediaType.JSON);
+ assertEquals(25, out.getCollection("exchangeEvents").size());
+ }
+
+ @Test
+ public void testSqlTraceCapacitySetAfterStart() {
+ SqlTraceDevConsole console = resolve("sql-trace");
+ console.setCapacity(1000);
+
+ // used to fail with ArrayIndexOutOfBoundsException as the dump walked
1000 slots of the default 200
+ JsonObject out = (JsonObject) console.call(DevConsole.MediaType.JSON);
+ assertTrue(out.isEmpty());
+ assertEquals("", console.call(DevConsole.MediaType.TEXT));
+ }
+
+ @Test
+ public void testInvalidCapacitySetAfterStart() {
+ // the range checked when the console starts is also checked when the
capacity is set afterwards
+ TraceDevConsole trace = resolve("trace");
+ assertInvalidCapacity(trace::getCapacity, trace::setCapacity, 50);
+ ReceiveDevConsole receive = resolve("receive");
+ assertInvalidCapacity(receive::getCapacity, receive::setCapacity, 50);
+ EventConsole event = resolve("event");
+ assertInvalidCapacity(event::getCapacity, event::setCapacity, 25);
+ SqlTraceDevConsole sqlTrace = resolve("sql-trace");
+ assertInvalidCapacity(sqlTrace::getCapacity, sqlTrace::setCapacity,
25);
+ }
+
+ private static void assertInvalidCapacity(IntSupplier getter, IntConsumer
setter, int min) {
+ int before = getter.getAsInt();
+ for (int capacity : new int[] { min - 1, 1001 }) {
+ IllegalArgumentException e =
assertThrows(IllegalArgumentException.class, () -> setter.accept(capacity));
+ assertEquals("Capacity must be between " + min + " and 1000",
e.getMessage());
+ assertEquals(before, getter.getAsInt());
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 8afd5488a35a..0eaa9c32cd09 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -645,6 +645,13 @@ The `inflight` and `blocked` developer consoles now carry
a `nodeSource` field p
exchange sits at is in the source (such as `orders.camel.yaml:18`). It is
`null` when message history or source
location is disabled. Existing fields are unchanged.
+The `capacity` option of the `trace`, `receive`, `event` and `sql-trace`
developer consoles, set from configuration
+(such as `camel.devConsole.trace.capacity`), is now applied. Prior to Camel
4.23 Camel Main set it after the console
+had started, so a capacity larger than the default broke the console (`Queue
full` or
+`ArrayIndexOutOfBoundsException`), and the documented range was not checked. A
capacity outside that range (50-1000
+for `trace` and `receive`, 25-1000 for `event` and `sql-trace`) now fails at
startup with an `IllegalArgumentException`,
+also when it is lower than the minimum, which was accepted before.
+
=== Error handler - redelivery options of onException
An `onException` that sets some of the redelivery options (such as
`maximumRedeliveries`) now inherits the options it
