From 0715b7ae78ee7941d1b8c2bf7ff6fb7251c314f2 Mon Sep 17 00:00:00 2001 From: Mitch Gaffigan Date: Tue, 27 Oct 2026 21:28:48 -0500 Subject: [PATCH 1/3] Add load testing to CI Signed-off-by: Mitch Gaffigan --- .../smoketest/ChainFanoutLoadTest.java | 165 ++++++++ .../smoketest/LoadSupport.java | 385 ++++++++++++++++++ .../smoketest/OieServer.java | 29 ++ .../smoketest/QueuedDestinationsLoadTest.java | 136 +++++++ .../test/resources/loadtest/load-control.xml | 272 +++++++++++++ .../test/resources/loadtest/load-entry.xml | 357 ++++++++++++++++ .../resources/loadtest/load-http-echo.xml | 188 +++++++++ .../resources/loadtest/load-mllp-echo.xml | 191 +++++++++ .../test/resources/loadtest/load-queue.xml | 356 ++++++++++++++++ 9 files changed, 2079 insertions(+) create mode 100644 smoketest/src/test/java/org/openintegrationengine/smoketest/ChainFanoutLoadTest.java create mode 100644 smoketest/src/test/java/org/openintegrationengine/smoketest/LoadSupport.java create mode 100644 smoketest/src/test/java/org/openintegrationengine/smoketest/QueuedDestinationsLoadTest.java create mode 100644 smoketest/src/test/resources/loadtest/load-control.xml create mode 100644 smoketest/src/test/resources/loadtest/load-entry.xml create mode 100644 smoketest/src/test/resources/loadtest/load-http-echo.xml create mode 100644 smoketest/src/test/resources/loadtest/load-mllp-echo.xml create mode 100644 smoketest/src/test/resources/loadtest/load-queue.xml diff --git a/smoketest/src/test/java/org/openintegrationengine/smoketest/ChainFanoutLoadTest.java b/smoketest/src/test/java/org/openintegrationengine/smoketest/ChainFanoutLoadTest.java new file mode 100644 index 000000000..84f58e29b --- /dev/null +++ b/smoketest/src/test/java/org/openintegrationengine/smoketest/ChainFanoutLoadTest.java @@ -0,0 +1,165 @@ +// SPDX-License-Identifier: MPL-2.0 +// SPDX-FileCopyrightText: 2026 Mitch Gaffigan + +package org.openintegrationengine.smoketest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.TestInstance.Lifecycle.PER_CLASS; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.TestReporter; + +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.model.message.Message; +import com.mirth.connect.model.ChannelStatistics; + +/** + * Pushes a thousand messages through a chain of three channels and checks every one of them. + * + *

The entry channel pulls each message off disk, pulls its blob out into an attachment, and + * routes it on content to exactly one of two loopback hops: MLLP to one channel, HTTP to + * another. Each hop answers with the SHA-256 of what actually arrived, and the control channel + * says up front what that digest and route should be. A match means the payload survived the + * attachment table and the hop byte for byte. + */ +@TestInstance(PER_CLASS) +@DisplayName("load: chained fan-out over MLLP and HTTP") +class ChainFanoutLoadTest { + + private static final int MESSAGES = 1000; + private static final String RUN = "entry"; + + /** Destination metadata ids, in the order the entry channel declares them. */ + private static final int MLLP = 1; + private static final int HTTP = 2; + + private String controlId; + private String entryId; + private String mllpEchoId; + private String httpEchoId; + private LoadSupport.Expectations expected; + + @BeforeAll + void deploy() throws Exception { + controlId = Harness.deploy(LoadSupport.CONTROL_CHANNEL); + + // Stage the whole run before anything is listening; arming it is the start of the run. + expected = LoadSupport.prepare(controlId, RUN, MESSAGES); + + mllpEchoId = Harness.deploy("loadtest/load-mllp-echo.xml"); + httpEchoId = Harness.deploy("loadtest/load-http-echo.xml"); + entryId = Harness.deploy("loadtest/load-entry.xml"); + } + + @AfterAll + void undeploy() throws Exception { + if (controlId != null) { + LoadSupport.clean(controlId); + } + Harness.undeploy(entryId); + Harness.undeploy(httpEchoId); + Harness.undeploy(mllpEchoId); + Harness.undeploy(controlId); + } + + @Test + @DisplayName(MESSAGES + " messages fan out to MLLP and HTTP with matching digests") + void fanOut(TestReporter reporter) throws Exception { + ChannelStatistics entryBefore = SharedServer.get().statistics(entryId); + long entryBaseline = entryBefore.getReceived(); + long entrySentBaseline = entryBefore.getSent(); + long entryErrorBaseline = entryBefore.getError(); + long mllpBaseline = SharedServer.get().statistics(mllpEchoId).getReceived(); + long httpBaseline = SharedServer.get().statistics(httpEchoId).getReceived(); + long mllpErrorBaseline = SharedServer.get().statistics(mllpEchoId).getError(); + long httpErrorBaseline = SharedServer.get().statistics(httpEchoId).getError(); + + // Nothing has been consumed yet, so starting the channel below is the one and only + // moment the engine can begin work. + SharedServer.get().stopChannel(entryId); + LoadSupport.arm(controlId, RUN); + + LoadSupport.Sample before = LoadSupport.stats(controlId); + long startNanos = System.nanoTime(); + SharedServer.get().startChannel(entryId); + long ingestNanos = LoadSupport.awaitReceived(entryId, entryBaseline, MESSAGES, startNanos); + + // Ingest only means the messages exist; the hops behind them still have to finish + // before every response is on record and worth asserting on. + LoadSupport.awaitDispatched(entryId, entrySentBaseline, entryErrorBaseline, MESSAGES); + LoadSupport.Sample after = LoadSupport.stats(controlId); + + List messages = LoadSupport.allMessages(entryId, MESSAGES); + assertEquals(MESSAGES, messages.size(), "messages stored on the entry channel"); + + List latencies = new ArrayList<>(MESSAGES); + boolean[] seen = new boolean[MESSAGES]; + int mllpCount = 0; + int httpCount = 0; + + for (Message message : messages) { + int seq = LoadSupport.sequenceOf(message); + assertTrue(seq >= 0 && seq < MESSAGES, "sequence " + seq + " is outside the run"); + assertTrue(!seen[seq], "sequence " + seq + " arrived more than once"); + seen[seq] = true; + + String route = expected.routes().get(seq); + boolean mllp = "MLLP".equals(route); + int taken = mllp ? MLLP : HTTP; + + ConnectorMessage destination = message.getConnectorMessages().get(taken); + assertNotNull(destination, "sequence " + seq + " should have been routed over " + route); + assertEquals(route + ":" + expected.digests().get(seq), + LoadSupport.responseText(destination.getResponse()), + "digest echoed back for sequence " + seq + " over " + route); + + // The routing step removes the other destination, so it must not have run at all. + assertNull(message.getConnectorMessages().get(mllp ? HTTP : MLLP), + "sequence " + seq + " also reached the destination it should have skipped"); + + if (mllp) { + mllpCount++; + } else { + httpCount++; + } + latencies.add(LoadSupport.latencyMillis(message)); + } + + assertEquals(MESSAGES, mllpCount + httpCount, "messages accounted for across both destinations"); + assertTrue(mllpCount > 0 && httpCount > 0, + "routing never varied: " + mllpCount + " over MLLP, " + httpCount + " over HTTP"); + + assertEquals(0, SharedServer.get().statistics(entryId).getError() - entryErrorBaseline, + "errors on the entry channel"); + assertEquals(mllpCount, SharedServer.get().statistics(mllpEchoId).getReceived() - mllpBaseline, + "messages received by the MLLP echo channel"); + assertEquals(httpCount, SharedServer.get().statistics(httpEchoId).getReceived() - httpBaseline, + "messages received by the HTTP echo channel"); + assertEquals(0, SharedServer.get().statistics(mllpEchoId).getError() - mllpErrorBaseline, + "errors on the MLLP echo channel"); + assertEquals(0, SharedServer.get().statistics(httpEchoId).getError() - httpErrorBaseline, + "errors on the HTTP echo channel"); + + new LoadSupport.Metrics("load: chained fan-out") + .put("messages", MESSAGES) + .put("bytes", expected.bytes()) + .put("routed.mllp", mllpCount) + .put("routed.http", httpCount) + .millis("ingest.millis", ingestNanos) + .rate("ingest.msgPerSec", MESSAGES, ingestNanos) + .rate("ingest.bytesPerSec", expected.bytes(), ingestNanos) + .latencies("latency.millis", latencies) + .resources(before, after, MESSAGES) + .publish(reporter); + } +} diff --git a/smoketest/src/test/java/org/openintegrationengine/smoketest/LoadSupport.java b/smoketest/src/test/java/org/openintegrationengine/smoketest/LoadSupport.java new file mode 100644 index 000000000..4221a70df --- /dev/null +++ b/smoketest/src/test/java/org/openintegrationengine/smoketest/LoadSupport.java @@ -0,0 +1,385 @@ +// SPDX-License-Identifier: MPL-2.0 +// SPDX-FileCopyrightText: 2026 Mitch Gaffigan + +package org.openintegrationengine.smoketest; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Calendar; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; + +import org.junit.jupiter.api.TestReporter; + +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.model.message.Message; +import com.mirth.connect.donkey.model.message.MessageContent; +import com.mirth.connect.donkey.model.message.Response; +import com.mirth.connect.model.ChannelStatistics; +import com.mirth.connect.model.converters.ObjectXMLSerializer; + +/** + * What the load tests need from the server, and nothing about the load itself. + * + *

{@code loadtest/load-control.xml} owns a run end to end: it writes the payload files + * inside the server container, knows where they live, and reports the digest and route each + * message should produce. The harness only issues commands and checks the answers against + * that report, so the payload rules live in one place and the harness never needs a view of + * the server's filesystem. + */ +final class LoadSupport { + + static final String CONTROL_CHANNEL = "loadtest/load-control.xml"; + + /** + * Ceiling on any one phase of a load run. The fixture tests' 90s is sized for a single + * message and is far too tight for a thousand of them on a slow database. + */ + private static final Duration TIMEOUT = Duration.ofMinutes(5); + + private static final int PAGE_SIZE = 100; + + /** Stored response content is a serialised {@link Response}, same as the fixture tests see. */ + private static final Pattern RESPONSE_ENVELOPE = Pattern.compile("^\\s*].*", Pattern.DOTALL); + + private LoadSupport() { + } + + // ------------------------------------------------------------- control channel + + /** + * Runs one control-channel command and returns its reply. The control channel answers + * before processing so a long command cannot outlive the client's request timeout, which + * is why this polls for the reply rather than reading it from the submit. + */ + static String control(String controlChannelId, String command, Map arguments) + throws Exception { + Map sourceMap = new LinkedHashMap<>(arguments); + sourceMap.put("cmd", command); + + List messageIds = SharedServer.get().submitMessage(controlChannelId, command, sourceMap); + if (messageIds.size() != 1) { + throw new AssertionError("Control command '" + command + "' produced " + messageIds.size() + + " messages, expected exactly one: " + messageIds); + } + long messageId = messageIds.get(0); + + long deadline = System.nanoTime() + TIMEOUT.toNanos(); + while (System.nanoTime() < deadline) { + Message message = SharedServer.get().fetchMessage(controlChannelId, messageId); + if (message != null && message.isProcessed()) { + ConnectorMessage connectorMessage = message.getConnectorMessages().get(1); + String reply = responseText(connectorMessage == null ? null : connectorMessage.getResponse()); + if (reply == null) { + throw new AssertionError("Control command '" + command + "' produced no response: " + + (connectorMessage == null ? "no connector message" + : connectorMessage.getStatus() + " " + connectorMessage.getProcessingError())); + } + return reply; + } + Thread.sleep(100); + } + throw new AssertionError("Control command '" + command + "' did not finish within " + + TIMEOUT.toSeconds() + "s"); + } + + /** + * What a prepared run should produce, as the control channel reports it: the digest each + * message's payload hashes to, and which loopback it should be routed over. + */ + record Expectations(List digests, List routes, long bytes) { + + int count() { + return digests.size(); + } + } + + /** + * Stages a run of {@code count} messages for {@code run} and returns what to expect. The + * files land in a staging directory, so nothing is visible to a File Reader until + * {@link #arm}. + */ + static Expectations prepare(String controlChannelId, String run, int count) throws Exception { + String reply = control(controlChannelId, "prepare", Map.of("run", run, "count", String.valueOf(count))); + + String[] lines = reply.split("\n"); + long bytes = Long.parseLong(lines[0].replaceFirst(".*;bytes=", "").trim()); + + List digests = new ArrayList<>(count); + List routes = new ArrayList<>(count); + for (int i = 1; i < lines.length; i++) { + String[] fields = lines[i].trim().split(","); + if (fields.length != 3) { + continue; + } + int seq = Integer.parseInt(fields[0]); + while (digests.size() <= seq) { + digests.add(null); + routes.add(null); + } + digests.set(seq, fields[1]); + routes.set(seq, fields[2]); + } + + if (digests.size() != count) { + throw new AssertionError("Control channel staged " + digests.size() + " of " + count + " messages"); + } + return new Expectations(digests, routes, bytes); + } + + /** Makes a prepared run visible to its File Reader, all at once. */ + static void arm(String controlChannelId, String run) throws Exception { + control(controlChannelId, "arm", Map.of("run", run)); + } + + static void clean(String controlChannelId) throws Exception { + control(controlChannelId, "clean", Map.of()); + } + + /** A snapshot of the server JVM, taken either side of a run. */ + record Sample(long cpuNanos, long heapUsed, long gcCount, long gcMillis, long threads) { + + static Sample parse(String reply) { + Map values = new LinkedHashMap<>(); + for (String pair : reply.trim().split(";")) { + String[] parts = pair.split("=", 2); + if (parts.length == 2) { + values.put(parts[0].trim(), Long.parseLong(parts[1].trim())); + } + } + return new Sample(values.getOrDefault("cpuNanos", -1L), values.getOrDefault("heapUsed", -1L), + values.getOrDefault("gcCount", -1L), values.getOrDefault("gcMillis", -1L), + values.getOrDefault("threads", -1L)); + } + } + + static Sample stats(String controlChannelId) throws Exception { + return Sample.parse(control(controlChannelId, "stats", Map.of())); + } + + // ------------------------------------------------------------------- polling + + /** + * Waits until the channel's lifetime received count has risen by {@code expected} and + * returns how long that took. Lifetime counters survive a redeploy, hence the baseline. + */ + static long awaitReceived(String channelId, long baseline, int expected, long startNanos) throws Exception { + long deadline = startNanos + TIMEOUT.toNanos(); + long received = 0; + while (System.nanoTime() < deadline) { + received = SharedServer.get().statistics(channelId).getReceived() - baseline; + if (received >= expected) { + return System.nanoTime() - startNanos; + } + Thread.sleep(50); + } + throw new AssertionError("Only " + received + " of " + expected + " messages reached channel " + channelId + + " within " + TIMEOUT.toSeconds() + "s"); + } + + /** + * Waits until every dispatch the run should produce has finished. The received counter + * rises when a message is created, not when its destinations are done, so on its own it + * lets a run be read back while the last messages are still in flight. + * + *

Errors count as finished: a run that errors should fail on the assertion that says + * so, not by sitting here until the timeout. + */ + static void awaitDispatched(String channelId, long sentBaseline, long errorBaseline, long expected) + throws Exception { + long deadline = System.nanoTime() + TIMEOUT.toNanos(); + long finished = 0; + while (System.nanoTime() < deadline) { + ChannelStatistics statistics = SharedServer.get().statistics(channelId); + finished = statistics.getSent() - sentBaseline + statistics.getError() - errorBaseline; + if (finished >= expected) { + return; + } + Thread.sleep(50); + } + throw new AssertionError("Only " + finished + " of " + expected + " dispatches finished on channel " + + channelId + " within " + TIMEOUT.toSeconds() + "s"); + } + + /** + * Waits for every destination queue to empty, returning the duration and the deepest + * queue seen. A max depth of zero means the queue drained as fast as it filled and the + * test is not exercising what it claims. + */ + static long[] awaitQueuesDrained(String channelId, long startNanos) throws Exception { + long deadline = startNanos + TIMEOUT.toNanos(); + long maxDepth = 0; + long depth = 0; + while (System.nanoTime() < deadline) { + depth = queueDepth(channelId); + maxDepth = Math.max(maxDepth, depth); + if (depth == 0) { + return new long[] { System.nanoTime() - startNanos, maxDepth }; + } + Thread.sleep(50); + } + throw new AssertionError("Queues on channel " + channelId + " still held " + depth + " messages after " + + TIMEOUT.toSeconds() + "s"); + } + + static long queueDepth(String channelId) throws Exception { + long depth = 0; + for (var child : SharedServer.get().status(channelId).getChildStatuses()) { + Long queued = child.getQueued(); + if (queued != null) { + depth += queued; + } + } + return depth; + } + + // ------------------------------------------------------------------ messages + + /** Reads every message on a channel, with content, so assertions can cover all of them. */ + static List allMessages(String channelId, int expected) throws Exception { + List messages = new ArrayList<>(expected); + for (int offset = 0; offset < expected; offset += PAGE_SIZE) { + List page = SharedServer.get().fetchMessages(channelId, true, offset, PAGE_SIZE); + if (page.isEmpty()) { + break; + } + messages.addAll(page); + } + return messages; + } + + /** The {@code seq=N} the control channel stamped on the payload. */ + static int sequenceOf(Message message) { + ConnectorMessage source = message.getConnectorMessages().get(0); + String raw = source == null ? null : text(source.getRaw()); + if (raw == null) { + throw new AssertionError("Message " + message.getMessageId() + " has no source content"); + } + + int start = raw.indexOf("seq=") + "seq=".length(); + return Integer.parseInt(raw.substring(start, raw.indexOf(';', start))); + } + + /** Null-safe content accessor; a connector may legitimately have no content of a kind. */ + static String text(MessageContent content) { + return content == null ? null : content.getContent(); + } + + /** The payload a connector answered with, unwrapped from its {@link Response} envelope. */ + static String responseText(MessageContent content) { + String stored = text(content); + if (stored == null || !RESPONSE_ENVELOPE.matcher(stored).matches()) { + return stored; + } + Response response = ObjectXMLSerializer.getInstance().deserialize(stored.trim(), Response.class); + return response == null ? null : response.getMessage(); + } + + /** + * End-to-end engine latency: from the message arriving to the last connector finishing. + * This excludes the harness and the network, so it stays comparable across machines. + */ + static long latencyMillis(Message message) { + Calendar received = message.getReceivedDate(); + if (received == null) { + return -1; + } + + long last = received.getTimeInMillis(); + for (ConnectorMessage connectorMessage : message.getConnectorMessages().values()) { + Calendar done = connectorMessage.getResponseDate() != null ? connectorMessage.getResponseDate() + : connectorMessage.getReceivedDate(); + if (done != null) { + last = Math.max(last, done.getTimeInMillis()); + } + } + return last - received.getTimeInMillis(); + } + + // ------------------------------------------------------------------- metrics + + /** + * An ordered set of measurements. Publishing sends each one to the JUnit report (report + * entries land in {@code } of the XML) and prints a table, because the XML is + * an artifact while the console output is right there in the CI job log. + */ + static final class Metrics { + + private final String title; + private final Map values = new LinkedHashMap<>(); + + Metrics(String title) { + this.title = title; + put("configuration", HarnessConfig.CONFIGURATION.isBlank() ? "(none)" : HarnessConfig.CONFIGURATION); + } + + Metrics put(String key, String value) { + values.put(key, value); + return this; + } + + Metrics put(String key, long value) { + return put(key, Long.toString(value)); + } + + Metrics rate(String key, long count, long nanos) { + return put(key, nanos <= 0 ? "n/a" : String.format("%.1f", count * 1_000_000_000.0 / nanos)); + } + + Metrics millis(String key, long nanos) { + return put(key, String.format("%.1f", nanos / 1_000_000.0)); + } + + /** Adds the usual spread for a set of per-message latencies, in milliseconds. */ + Metrics latencies(String prefix, List samples) { + if (samples.isEmpty()) { + return put(prefix, "n/a"); + } + List sorted = new ArrayList<>(samples); + sorted.sort(null); + + long sum = 0; + for (long sample : sorted) { + sum += sample; + } + put(prefix + ".mean", String.format("%.1f", (double) sum / sorted.size())); + put(prefix + ".p50", percentile(sorted, 0.50)); + put(prefix + ".p90", percentile(sorted, 0.90)); + put(prefix + ".p95", percentile(sorted, 0.95)); + put(prefix + ".p99", percentile(sorted, 0.99)); + put(prefix + ".max", sorted.get(sorted.size() - 1)); + return this; + } + + /** The JVM counters either side of a run, as deltas. */ + Metrics resources(Sample before, Sample after, int messages) { + if (before.cpuNanos() < 0 || after.cpuNanos() < 0) { + put("cpu.seconds", "n/a").put("cpu.millisPerMessage", "n/a"); + } else { + long cpuNanos = after.cpuNanos() - before.cpuNanos(); + put("cpu.seconds", String.format("%.3f", cpuNanos / 1_000_000_000.0)); + put("cpu.millisPerMessage", String.format("%.3f", cpuNanos / 1_000_000.0 / messages)); + } + return put("gc.count", after.gcCount() - before.gcCount()) + .put("gc.millis", after.gcMillis() - before.gcMillis()) + .put("heap.usedDelta", after.heapUsed() - before.heapUsed()) + .put("threads", after.threads()); + } + + private static long percentile(List sorted, double fraction) { + int index = (int) Math.ceil(fraction * sorted.size()) - 1; + return sorted.get(Math.min(Math.max(index, 0), sorted.size() - 1)); + } + + void publish(TestReporter reporter) { + reporter.publishEntry(values); + + StringBuilder table = new StringBuilder("\n").append(title).append('\n'); + int width = values.keySet().stream().mapToInt(String::length).max().orElse(1); + values.forEach((key, value) -> table.append(String.format(" %-" + width + "s %s%n", key, value))); + System.out.print(table); + } + } +} diff --git a/smoketest/src/test/java/org/openintegrationengine/smoketest/OieServer.java b/smoketest/src/test/java/org/openintegrationengine/smoketest/OieServer.java index 3fd16b042..3275852b0 100644 --- a/smoketest/src/test/java/org/openintegrationengine/smoketest/OieServer.java +++ b/smoketest/src/test/java/org/openintegrationengine/smoketest/OieServer.java @@ -15,6 +15,8 @@ import com.mirth.connect.donkey.model.message.Message; import com.mirth.connect.donkey.model.message.RawMessage; import com.mirth.connect.model.Channel; +import com.mirth.connect.model.ChannelStatistics; +import com.mirth.connect.model.DashboardStatus; import com.mirth.connect.model.LoginStatus; import com.mirth.connect.model.converters.ObjectXMLSerializer; import com.mirth.connect.model.filters.MessageFilter; @@ -145,6 +147,33 @@ List submitMessage(String channelId, String rawData, Map s return messageIds; } + /** Stops a deployed channel, so a load test can line its input up before anything runs. */ + void stopChannel(String channelId) throws ClientException { + client.stopChannel(channelId, true); + } + + /** Starts a stopped channel. For a polling source this is the start of the run. */ + void startChannel(String channelId) throws ClientException { + client.startChannel(channelId, true); + } + + /** Lifetime received/sent/error/filtered/queued counts for a channel. */ + ChannelStatistics statistics(String channelId) throws ClientException { + return client.getStatistics(channelId); + } + + /** Dashboard view of a channel, including live queue depth per destination. */ + DashboardStatus status(String channelId) throws ClientException { + return client.getChannelStatus(channelId); + } + + /** Reads a page of messages, newest id last, for bulk assertions after a load run. */ + List fetchMessages(String channelId, boolean includeContent, int offset, int limit) + throws ClientException { + List messages = client.getMessages(channelId, new MessageFilter(), includeContent, offset, limit); + return messages == null ? List.of() : messages; + } + /** Reads one message back, with content, so assertions can inspect every connector. */ Message fetchMessage(String channelId, long messageId) throws ClientException { MessageFilter filter = new MessageFilter(); diff --git a/smoketest/src/test/java/org/openintegrationengine/smoketest/QueuedDestinationsLoadTest.java b/smoketest/src/test/java/org/openintegrationengine/smoketest/QueuedDestinationsLoadTest.java new file mode 100644 index 000000000..12c2b905a --- /dev/null +++ b/smoketest/src/test/java/org/openintegrationengine/smoketest/QueuedDestinationsLoadTest.java @@ -0,0 +1,136 @@ +// SPDX-License-Identifier: MPL-2.0 +// SPDX-FileCopyrightText: 2026 Mitch Gaffigan + +package org.openintegrationengine.smoketest; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.TestInstance.Lifecycle.PER_CLASS; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.TestReporter; + +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.model.message.Message; +import com.mirth.connect.donkey.model.message.Status; +import com.mirth.connect.model.ChannelStatistics; + +/** + * The simpler half: one channel, three independent queued destinations, four queue readers + * each. Nothing chains onwards, so what this exercises is the queue itself — filling it faster + * than it drains, then draining it with several readers per destination. + * + *

Ingest and drain are timed separately because the source answers as soon as a message is + * accepted; the work carries on behind it. + */ +@TestInstance(PER_CLASS) +@DisplayName("load: queued destinations with multiple readers") +class QueuedDestinationsLoadTest { + + private static final int MESSAGES = 1000; + private static final int DESTINATIONS = 3; + private static final String RUN = "queue"; + + private String controlId; + private String queueId; + private LoadSupport.Expectations expected; + + @BeforeAll + void deploy() throws Exception { + controlId = Harness.deploy(LoadSupport.CONTROL_CHANNEL); + expected = LoadSupport.prepare(controlId, RUN, MESSAGES); + queueId = Harness.deploy("loadtest/load-queue.xml"); + } + + @AfterAll + void undeploy() throws Exception { + if (controlId != null) { + LoadSupport.clean(controlId); + } + Harness.undeploy(queueId); + Harness.undeploy(controlId); + } + + @Test + @DisplayName(MESSAGES + " messages through " + DESTINATIONS + " queues, four readers each") + void drainQueues(TestReporter reporter) throws Exception { + ChannelStatistics startStats = SharedServer.get().statistics(queueId); + long baseline = startStats.getReceived(); + long sentBaseline = startStats.getSent(); + long errorBaseline = startStats.getError(); + + SharedServer.get().stopChannel(queueId); + LoadSupport.arm(controlId, RUN); + + LoadSupport.Sample before = LoadSupport.stats(controlId); + long startNanos = System.nanoTime(); + SharedServer.get().startChannel(queueId); + + long ingestNanos = LoadSupport.awaitReceived(queueId, baseline, MESSAGES, startNanos); + long[] drain = LoadSupport.awaitQueuesDrained(queueId, System.nanoTime()); + + // An empty queue means nothing is waiting, not that the last dispatches off it have + // finished recording their responses. + LoadSupport.awaitDispatched(queueId, sentBaseline, errorBaseline, (long) MESSAGES * DESTINATIONS); + LoadSupport.Sample after = LoadSupport.stats(controlId); + + long drainNanos = drain[0]; + long maxQueueDepth = drain[1]; + + List messages = LoadSupport.allMessages(queueId, MESSAGES); + assertEquals(MESSAGES, messages.size(), "messages stored on the queue channel"); + + List latencies = new ArrayList<>(MESSAGES); + boolean[] seen = new boolean[MESSAGES]; + + for (Message message : messages) { + int seq = LoadSupport.sequenceOf(message); + assertTrue(seq >= 0 && seq < MESSAGES, "sequence " + seq + " is outside the run"); + assertTrue(!seen[seq], "sequence " + seq + " arrived more than once"); + seen[seq] = true; + + String digest = expected.digests().get(seq); + for (int metaDataId = 1; metaDataId <= DESTINATIONS; metaDataId++) { + ConnectorMessage destination = message.getConnectorMessages().get(metaDataId); + assertNotNull(destination, "sequence " + seq + " never reached destination " + metaDataId); + assertEquals(Status.SENT, destination.getStatus(), + "status of destination " + metaDataId + " for sequence " + seq); + assertEquals("D" + metaDataId + ":" + digest, + LoadSupport.responseText(destination.getResponse()), + "digest from destination " + metaDataId + " for sequence " + seq); + } + latencies.add(LoadSupport.latencyMillis(message)); + } + + assertEquals(0, LoadSupport.queueDepth(queueId), "queue depth once the run has settled"); + assertEquals(0, SharedServer.get().statistics(queueId).getError() - errorBaseline, + "errors on the queue channel"); + assertEquals((long) MESSAGES * DESTINATIONS, SharedServer.get().statistics(queueId).getSent() - sentBaseline, + "messages sent across all destinations"); + + // A queue that never backs up is not being exercised; say so rather than passing quietly. + assertTrue(maxQueueDepth > 0, "destination queues never backed up, so queueing went untested"); + + new LoadSupport.Metrics("load: queued destinations") + .put("messages", MESSAGES) + .put("bytes", expected.bytes()) + .put("destinations", DESTINATIONS) + .put("queue.maxDepth", maxQueueDepth) + .millis("ingest.millis", ingestNanos) + .rate("ingest.msgPerSec", MESSAGES, ingestNanos) + .rate("ingest.bytesPerSec", expected.bytes(), ingestNanos) + .millis("drain.millis", drainNanos) + .rate("drain.dispatchPerSec", (long) MESSAGES * DESTINATIONS, drainNanos) + .latencies("latency.millis", latencies) + .resources(before, after, MESSAGES) + .publish(reporter); + } +} diff --git a/smoketest/src/test/resources/loadtest/load-control.xml b/smoketest/src/test/resources/loadtest/load-control.xml new file mode 100644 index 000000000..d055acdce --- /dev/null +++ b/smoketest/src/test/resources/loadtest/load-control.xml @@ -0,0 +1,272 @@ + + 8f1c6a40-0001-4a00-9a00-0000000000c0 + 2 + Load Control + Smoke-test support channel: writes load payloads and samples JVM counters. + 1 + + 0 + sourceConnector + + + None + false + false + false + 1 + + + Default Resource + [Default Resource] + + + 1000 + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + Channel Reader + SOURCE + true + true + + + + 1 + Control + + + false + false + 10000 + false + 0 + false + false + 1 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + JavaScript Writer + DESTINATION + true + true + + + return message; + return; + return; + return; + + true + PRODUCTION + false + false + false + false + false + false + STARTED + true + + + None + + + + + Default Resource + [Default Resource] + + + + diff --git a/smoketest/src/test/resources/loadtest/load-entry.xml b/smoketest/src/test/resources/loadtest/load-entry.xml new file mode 100644 index 000000000..39c2ff6e1 --- /dev/null +++ b/smoketest/src/test/resources/loadtest/load-entry.xml @@ -0,0 +1,357 @@ + + 8f1c6a40-0001-4a00-9a00-0000000000c1 + 3 + Load Entry + Smoke-test load fixture: fans out to the MLLP and HTTP echo channels. + 1 + + 0 + sourceConnector + + + INTERVAL + true + 100 + 0 + 0 + + + true + + false + false + false + false + false + false + false + false + + 1 + true + 8 + 0 + 17 + 0 + + + + None + true + false + false + 1 + + + Default Resource + [Default Resource] + + + 1000 + + FILE + /tmp/oie-load/entry + *.txt + false + false + true + true + anonymous + anonymous + 10000 + true + true + true + DELETE + + + NONE + AFTER_PROCESSING + + + false + 1000 + 0 + + true + name + false + UTF-8 + + + + + Route on content + 0 + true + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + File Reader + SOURCE + true + true + + + + 1 + MLLP Loopback + + + false + false + 10000 + false + 0 + false + false + 1 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + MLLP + 0B + 1C0D + + false + 127.0.0.1 + 9082 + false + 0.0.0.0 + 0 + 30000 + 65536 + 64 + true + false + 30000 + false + true + false + DEFAULT_ENCODING + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + TCP Sender + DESTINATION + true + true + + + 2 + HTTP Loopback + + + false + false + 10000 + false + 0 + false + false + 1 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + http://127.0.0.1:9083/echo/ + false + + + false + 0.0.0.0 + post + + + false + + false + + false + true + false + application/.*(?<!json|xml)$|image/.*|video/.*|audio/.* + true + false + false + Basic + false + + + ${message.encodedData} + text/plain + false + UTF-8 + 30000 + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + HTTP Sender + DESTINATION + true + true + + + return message; + return; + return; + return; + + true + PRODUCTION + false + false + false + false + false + false + STARTED + true + + + com.mirth.connect.server.attachments.regex.RegexAttachmentHandlerProvider + Regex + + + regex.pattern0 + <BLOB>(.*?)</BLOB> + + + regex.mimetype0 + text/plain + + + + + + Default Resource + [Default Resource] + + + + diff --git a/smoketest/src/test/resources/loadtest/load-http-echo.xml b/smoketest/src/test/resources/loadtest/load-http-echo.xml new file mode 100644 index 000000000..f4c1173f3 --- /dev/null +++ b/smoketest/src/test/resources/loadtest/load-http-echo.xml @@ -0,0 +1,188 @@ + + 8f1c6a40-0001-4a00-9a00-0000000000c3 + 2 + Load HTTP Echo + Smoke-test load fixture: answers an HTTP request with its SHA-256. + 1 + + 0 + sourceConnector + + + 0.0.0.0 + 9083 + + + Postprocessor + true + false + false + 1 + + + Default Resource + [Default Resource] + + + 1000 + + false + false + false + application/.*(?<!json|xml)$|image/.*|video/.*|audio/.* + true + text/plain + false + + + + false + UTF-8 + /echo + 30000 + 8192 + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + HTTP Listener + SOURCE + true + true + + + + 1 + Noop + + + false + false + 10000 + false + 0 + false + false + 1 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + JavaScript Writer + DESTINATION + true + true + + + return message; + // Answer with a digest of exactly what arrived, so the caller can prove the +// payload survived the hop unchanged. +var raw = message.getMergedConnectorMessage().getRawData(); +return 'HTTP:' + HashUtil.generate(raw, 'UTF-8', 'SHA-256'); + return; + return; + + true + PRODUCTION + false + false + false + false + false + false + STARTED + true + + + None + + + + + Default Resource + [Default Resource] + + + + diff --git a/smoketest/src/test/resources/loadtest/load-mllp-echo.xml b/smoketest/src/test/resources/loadtest/load-mllp-echo.xml new file mode 100644 index 000000000..b06b01055 --- /dev/null +++ b/smoketest/src/test/resources/loadtest/load-mllp-echo.xml @@ -0,0 +1,191 @@ + + 8f1c6a40-0001-4a00-9a00-0000000000c2 + 2 + Load MLLP Echo + Smoke-test load fixture: answers an MLLP message with its SHA-256. + 1 + + 0 + sourceConnector + + + 0.0.0.0 + 9082 + + + Postprocessor + true + false + true + 1 + + + Default Resource + [Default Resource] + + + 1000 + + + MLLP + 0B + 1C0D + + true + + + false + 5000 + 0 + 65536 + 64 + true + false + DEFAULT_ENCODING + 0 + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + TCP Listener + SOURCE + true + true + + + + 1 + Noop + + + false + false + 10000 + false + 0 + false + false + 1 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + JavaScript Writer + DESTINATION + true + true + + + return message; + // Answer with a digest of exactly what arrived, so the caller can prove the +// payload survived the hop unchanged. +var raw = message.getMergedConnectorMessage().getRawData(); +return 'MLLP:' + HashUtil.generate(raw, 'UTF-8', 'SHA-256'); + return; + return; + + true + PRODUCTION + false + false + false + false + false + false + STARTED + true + + + None + + + + + Default Resource + [Default Resource] + + + + diff --git a/smoketest/src/test/resources/loadtest/load-queue.xml b/smoketest/src/test/resources/loadtest/load-queue.xml new file mode 100644 index 000000000..688d7f7ef --- /dev/null +++ b/smoketest/src/test/resources/loadtest/load-queue.xml @@ -0,0 +1,356 @@ + + 8f1c6a40-0001-4a00-9a00-0000000000c4 + 4 + Load Queue + Smoke-test load fixture: three queued destinations with four readers each. + 1 + + 0 + sourceConnector + + + INTERVAL + true + 100 + 0 + 0 + + + true + + false + false + false + false + false + false + false + false + + 1 + true + 8 + 0 + 17 + 0 + + + + None + true + false + false + 1 + + + Default Resource + [Default Resource] + + + 1000 + + FILE + /tmp/oie-load/queue + *.txt + false + false + true + true + anonymous + anonymous + 10000 + true + true + true + DELETE + + + NONE + AFTER_PROCESSING + + + false + 1000 + 0 + + true + name + false + UTF-8 + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + File Reader + SOURCE + true + true + + + + 1 + Queued 1 + + + true + false + 10000 + false + 0 + false + false + 4 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + JavaScript Writer + DESTINATION + true + false + + + 2 + Queued 2 + + + true + false + 10000 + false + 0 + false + false + 4 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + JavaScript Writer + DESTINATION + true + false + + + 3 + Queued 3 + + + true + false + 10000 + false + 0 + false + false + 4 + + false + + + Default Resource + [Default Resource] + + + 1000 + true + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + RAW + RAW + + + JavaScript + + + + + + JavaScript + + + + + + + + JavaScript Writer + DESTINATION + true + false + + + return message; + return; + return; + return; + + true + PRODUCTION + false + false + false + false + false + false + STARTED + true + + + None + + + + + Default Resource + [Default Resource] + + + + From 0fecb6c6f76f8f6a601045bafb408198f6973ea1 Mon Sep 17 00:00:00 2001 From: Mitch Gaffigan Date: Tue, 27 Oct 2026 23:06:20 -0500 Subject: [PATCH 2/3] Add OIE server log to CI runs that fail Signed-off-by: Mitch Gaffigan --- ci/run-configuration.sh | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/ci/run-configuration.sh b/ci/run-configuration.sh index 7881f57bb..6c04351f0 100755 --- a/ci/run-configuration.sh +++ b/ci/run-configuration.sh @@ -37,4 +37,10 @@ if ! "${compose[@]}" up -d --wait oie; then fi echo "==> $OIE_CONFIGURATION: running the harness" -"${compose[@]}" run --rm harness \ No newline at end of file +"${compose[@]}" run --rm harness || { + # Harness failed, dump the server log before exiting + code=$? + echo "==> $OIE_CONFIGURATION: harness failed with exit code $code" >&2 + "${compose[@]}" logs --no-color oie >&2 || true + exit "$code" +} \ No newline at end of file From 4ba704988526a8b2a8acd342637a800bd463d51d Mon Sep 17 00:00:00 2001 From: Mitch Gaffigan Date: Tue, 27 Oct 2026 23:39:07 -0500 Subject: [PATCH 3/3] Add logging for SQLException - currently opaque Signed-off-by: Mitch Gaffigan --- .../MirthResourceInvocationHandlerProvider.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/server/src/main/java/com/mirth/connect/server/api/providers/MirthResourceInvocationHandlerProvider.java b/server/src/main/java/com/mirth/connect/server/api/providers/MirthResourceInvocationHandlerProvider.java index 7f9688b3c..c359cb193 100644 --- a/server/src/main/java/com/mirth/connect/server/api/providers/MirthResourceInvocationHandlerProvider.java +++ b/server/src/main/java/com/mirth/connect/server/api/providers/MirthResourceInvocationHandlerProvider.java @@ -13,6 +13,7 @@ import java.lang.reflect.InvocationHandler; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.sql.SQLException; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -26,6 +27,8 @@ import org.apache.commons.collections4.CollectionUtils; import org.apache.ibatis.exceptions.PersistenceException; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.glassfish.jersey.server.model.Invocable; import org.glassfish.jersey.server.spi.internal.ResourceMethodInvocationHandlerProvider; @@ -42,6 +45,8 @@ @Provider public class MirthResourceInvocationHandlerProvider implements ResourceMethodInvocationHandlerProvider { + private static final Logger logger = LogManager.getLogger(MirthResourceInvocationHandlerProvider.class); + /* * This map is used to cache information about specific methods so we don't have to calculate it * every time. @@ -243,7 +248,11 @@ private Throwable convertThrowable(Throwable t, Set visited) { Throwable cause = t.getCause(); Throwable convertedCause = convertThrowable(cause, visited); - if (t instanceof PersistenceException) { + if (t instanceof SQLException) { + // Not serializable by XStream due to module restrictions, so log and wrap it instead + logger.error("Database error while handling an API request", t); + return new com.mirth.connect.client.core.api.PersistenceException(t.getMessage(), convertedCause); + } else if (t instanceof PersistenceException) { // Always convert this exception return new com.mirth.connect.client.core.api.PersistenceException(t.getMessage(), convertedCause != null ? convertedCause : cause); } else if (t instanceof WebApplicationException) {