[
https://issues.apache.org/jira/browse/ARTEMIS-5573?focusedWorklogId=1012965&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-1012965
]
ASF GitHub Bot logged work on ARTEMIS-5573:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 02/Apr/26 13:26
Start Date: 02/Apr/26 13:26
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #6323:
URL: https://github.com/apache/artemis/pull/6323#discussion_r3028042962
##########
tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/AMQPGlobalMaxTest.java:
##########
@@ -0,0 +1,356 @@
+/*
+ * 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.activemq.artemis.tests.soak.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import java.io.File;
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+
+import org.apache.activemq.artemis.api.core.management.SimpleManagement;
+import org.apache.activemq.artemis.cli.commands.helper.HelperCreate;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.apache.activemq.artemis.utils.FileUtil;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.Wait;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+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.Assertions.fail;
+
+public class AMQPGlobalMaxTest extends SoakTestBase {
+
+ private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final String QUEUE_NAME = "simpleTest";
+ public static final String SERVER_NAME = "global-max-test";
+ private static File serverLocation;
+
+ private int lastMessageSent;
+
+ String memoryUsed;
+ long highestMemoryUsed;
+
+ boolean hasOME = false;
+
+ private Process server;
+
+ private long avgSizeFirstEstimate;
+
+ public static void createServers(long globalMaxSize, String vmSize) throws
Exception {
+ serverLocation = getFileServerLocation(SERVER_NAME);
+ deleteDirectory(serverLocation);
+
+ HelperCreate cliCreateServer = helperCreate();
+
cliCreateServer.setUseAIO(false).setAllowAnonymous(true).setNoWeb(true).setArtemisInstance(serverLocation);
+ // to speedup producers
+ cliCreateServer.addArgs("--no-fsync");
+ cliCreateServer.addArgs("--queues", QUEUE_NAME);
+ cliCreateServer.addArgs("--global-max-size",
String.valueOf(globalMaxSize));
+ // limiting memory to make the test more predictable
+ cliCreateServer.addArgs("--java-memory", vmSize);
+ cliCreateServer.createServer();
+
+ FileUtil.findReplace(new File(serverLocation, "/etc/artemis.profile"),
"-Xms512M", "-Xms10M -verbose:gc");
+ }
+
+ @Test
+ public void testValidateMemoryEstimateLargeAMQP() throws Exception {
+ validateOME((s, i) -> {
+ try {
+ Message m = s.createTextMessage("a".repeat(200 * 1024));
+ m.setStringProperty("prop", "a".repeat(10 * 1024));
+ m.setStringProperty("prop", "b".repeat(10 * 1024));
+ return m;
+ } catch (Exception e) {
+ fail(e.getMessage());
+ return null;
+ }
+ }, "35M", 1, 100);
+ }
+
+ private void validateOME(BiFunction<Session, Integer, Message>
messageCreator, String vmSize, int commitInterval, int printInterval) throws
Exception {
+ // making the broker to OME on purpose
+ // this is to help us calculate the size of each message
+ // for this reason the global-max-size is set really high on this test
+ createServers(1024 * 1024 * 1024, vmSize);
+ startServerWithLog();
+
+ executeTest(messageCreator, (a, b) -> { }, vmSize, commitInterval,
printInterval, 20_000_000, TimeUnit.MINUTES.toMillis(10), true);
+ }
+
+ @Test
+ public void testValidateMemoryEstimateAMQP() throws Exception {
+ validateOME((s, i) -> {
+ try {
+ Message m = s.createMessage();
+ for (int randomI = 0; randomI < 10; randomI++) {
+ m.setStringProperty("string" + i,
RandomUtil.randomUUIDString());
+ m.setLongProperty("myLong" + i, RandomUtil.randomLong());
+ }
+ return m;
+ } catch (Throwable e) {
+ Assertions.fail(e.getMessage());
+ return null;
+ }
+ }, "256M", 1000, 1000);
+ }
+
+ private void startServerWithLog() throws Exception {
+ server = startServer(SERVER_NAME, 0, 5000, null, s -> {
+ logger.debug("{}", s);
+ if (s.contains("GC") && s.contains("->")) {
+ AMQPGlobalMaxTest.this.memoryUsed = parseMemoryUsageFromGCLOG(s);
+ long memoryUsedBytes = parseMemoryToBytes(memoryUsed);
+ if (memoryUsedBytes > highestMemoryUsed) {
+ highestMemoryUsed = memoryUsedBytes;
+ logger.info("Currently using {} on the server", memoryUsed);
+ }
+ }
+
+ // Stop Page, start page or anything important
+ if (s.contains("INFO") || s.contains("WARN")) {
+ logger.info("{}", s);
+ }
+
+ if (s.contains("OutOfMemoryError")) {
+
logger.info(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
OME!!!");
+ AMQPGlobalMaxTest.this.hasOME = true;
+ }
+ });
+ }
+
+ @Test
+ public void testGlobalMax() throws Exception {
+ createServers(100 * 1024 * 1024, "256M");
+ startServerWithLog();
+
+ executeTest((s, i) -> {
+ try {
+ Message message;
+ if (i % 100 == 0) {
+ // 10 large messages per transaction
+ message = s.createTextMessage("a".repeat(101 * 1024));
+ message.setStringProperty("i", "a".repeat(10 * 1024));
+ } else {
+ message = s.createMessage();
+ }
+ message.setIntProperty("i", i);
+ message.setStringProperty("someString", "a".repeat(1024));
+ return message;
+ } catch (Throwable e) {
+ Assertions.fail(e.getMessage());
+ return null;
+ }
+ }, (i, m) -> {
+ try {
+ assertEquals(i, m.getIntProperty("i"));
+ } catch (Throwable e) {
+ Assertions.fail(e.getMessage());
+ }
+ }, "256M", 10_000, 10_000, 250_000, TimeUnit.MINUTES.toMillis(10),
false);
+ }
+
+ private void executeTest(BiFunction<Session, Integer, Message>
messageCreator,
+ BiConsumer<Integer, Message> messageVerifier,
+ String memoryAllocated,
+ int commitInterval,
+ int printInterval,
+ int maxMessages,
+ long timeoutMilliseconds,
+ boolean expectOME) throws Exception {
+ ExecutorService service = Executors.newFixedThreadPool(2);
+ runAfter(service::shutdownNow);
+
+ CountDownLatch latchDone = new CountDownLatch(1);
+ AtomicInteger errors = new AtomicInteger(0);
+
+ // use some management operation to parse the properties
+ service.execute(() -> {
+ try {
+ SimpleManagement simpleManagement = new
SimpleManagement("tcp://localhost:61616", null, null);
+ while (!latchDone.await(1, TimeUnit.SECONDS)) {
+ // this filter will not remove any messages, but it will make
the messages to parse the application properties
Review Comment:
different contexts I guess... by parse here I mean.. they are decoded and
will use memory after this use.
Issue Time Tracking
-------------------
Worklog Id: (was: 1012965)
Time Spent: 50m (was: 40m)
> Make AMQP Size immutable
> ------------------------
>
> Key: ARTEMIS-5573
> URL: https://issues.apache.org/jira/browse/ARTEMIS-5573
> Project: Artemis
> Issue Type: Improvement
> Components: AMQP
> Affects Versions: 2.41.0
> Reporter: Clebert Suconic
> Assignee: Clebert Suconic
> Priority: Major
> Labels: pull-request-available
> Time Spent: 50m
> Remaining Estimate: 0h
>
> I have had a lot of issues, even recently on races between re-evaluating a
> message size in AMQP.
> Say a lazy decode happens at the wrong time and the memory estimates can be
> wrong.
> We have fixed issues along the years, but this is still a fragile process
> that is bound to fail. If an user for instance add a plugin breaking the
> chain of events.
> For that reason the memory estimate should already include enough estimation
> for any properties decoded and the process should be simplified.
> Less moving parts would mean less possibilities for bugs.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
