You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by "gaohoward (via GitHub)" <gi...@apache.org> on 2024/08/20 03:10:09 UTC
[PR] ENTMQBR-9338 AMQP producer not unblock if the disk space is freed [activemq-artemis]
gaohoward opened a new pull request, #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160
(test adapted from the one on the Jira)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic closed pull request #5160: ARTEMIS-5002 AMQP producer not unblock if the disk space is freed
URL: https://github.com/apache/activemq-artemis/pull/5160
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#issuecomment-2301511236
@clebertsuconic @gemmellr I've updated the PR. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1723914615
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ this.addAdditionalAcceptors(server);
+ this.configureAddressPolicy(server);
+ this.configureBrokerSecurity(server);
+ this.addConfiguration(server);
+ server.start();
+ this.createAddressAndQueues(server);
+ return server;
+ }
+
+ @Test
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ threadWithName.start();
+
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
+ anonSender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ sentAnon.countDown();
+ }
+ });
+
+ threadWithAnon.start();
Review Comment:
+1 on the executor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1724824633
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullFailPolicyTest.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class GlobalDiskFullFailPolicyTest extends GlobalDiskFullTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameter(index = 0)
+ public AddressFullMessagePolicy addressFullPolicy;
+
+ @Parameters(name = "addressFullPolicy={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {AddressFullMessagePolicy.FAIL}, {AddressFullMessagePolicy.DROP}
+ });
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(addressFullPolicy);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ }
+
+ @TestTemplate
+ @Override
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ logger.warn("Caught exception while sending", e);
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
Review Comment:
This uses getQueueName() but the other sender (and the address settings) use "TEST", meaning the policy isnt being correctly applied in this case.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullFailPolicyTest.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class GlobalDiskFullFailPolicyTest extends GlobalDiskFullTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameter(index = 0)
+ public AddressFullMessagePolicy addressFullPolicy;
+
+ @Parameters(name = "addressFullPolicy={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {AddressFullMessagePolicy.FAIL}, {AddressFullMessagePolicy.DROP}
+ });
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(addressFullPolicy);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ }
+
+ @TestTemplate
+ @Override
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ logger.warn("Caught exception while sending", e);
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
+ anonSender.send(message);
+ sentAnon.countDown();
+ } catch (Exception e) {
+ logger.warn("Caught exception while sending", e);
+ }
+ });
+
+ ExecutorService pool = Executors.newCachedThreadPool();
+ runAfter(pool::shutdownNow);
+
+ pool.execute(threadWithName);
+ pool.execute(threadWithAnon);
+
+ assertFalse(sentWithName.await(500, TimeUnit.MILLISECONDS), "Thread sender should be blocked");
+ assertFalse(sentAnon.await(500, TimeUnit.MILLISECONDS), "Thread sender anonymous should be blocked");
+ monitor.setMaxUsage(100.0);
+
+ assertTrue(sentWithName.await(30, TimeUnit.SECONDS), "Thread sender should be released");
+ assertTrue(sentAnon.await(30, TimeUnit.SECONDS), "Thread sender anonymous should be released");
+
Review Comment:
I would drop the newline at the end, and put one before the `monitor.setMaxUsage(100.0);` to help set it apart from the assertions before it, helping draw attention to the key step of the test, which changes the expected behaviour for the assertions following.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullFailPolicyTest.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class GlobalDiskFullFailPolicyTest extends GlobalDiskFullTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameter(index = 0)
+ public AddressFullMessagePolicy addressFullPolicy;
+
+ @Parameters(name = "addressFullPolicy={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {AddressFullMessagePolicy.FAIL}, {AddressFullMessagePolicy.DROP}
+ });
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(addressFullPolicy);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ }
+
+ @TestTemplate
+ @Override
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
Review Comment:
This is only checking that the callback is fired at all, which doesnt necessarily mean it was correct. This needs to verify that the callback indicates the desired state and type before it trips the latch.
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -1096,6 +1096,7 @@ public boolean checkMemory(boolean runOnFailure, Runnable runWhenAvailableParame
if (isFull()) {
if (runOnFailure && runWhenAvailable != null) {
addToBlockList(runWhenAvailable, blockedCallback);
+ pagingManager.addBlockedStore(this);
Review Comment:
The use of this same call lower down the class is gated by `if (usingGlobalMaxSize || pagingManager.isDiskFull()) {``, does this need to be?
It is also surrounded by various other double-checks around racing being not-full again, it seems hard to reconcile that path needing it but this one not.
I cant quite wrap my head around this so think @clebertsuconic might need to chip in here.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullFailPolicyTest.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class GlobalDiskFullFailPolicyTest extends GlobalDiskFullTest {
Review Comment:
It really doesnt make sense for this class to subclass GlobalDiskFullTest and then _override the only test_ in it _with effectively the same test_, and also convert from non-parameterized to parameterized. GlobalDiskFullFailPolicyTest also doesnt really work as a name when it is testing both FAIL _and_ DROP.
My other previous feedback was that if this general test basically works for all 3 policies, then we could essentially remove the test code from this class (or the whole class) entirely and update GlobalDiskFullTest to have that test method cover all 3 variants, either by directly parameterizing that class to set all the policies (as you did here for fail+drop) or by having subclasses that basically just pass the policy value.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullFailPolicyTest.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class GlobalDiskFullFailPolicyTest extends GlobalDiskFullTest {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameter(index = 0)
+ public AddressFullMessagePolicy addressFullPolicy;
+
+ @Parameters(name = "addressFullPolicy={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {AddressFullMessagePolicy.FAIL}, {AddressFullMessagePolicy.DROP}
+ });
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(addressFullPolicy);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ }
+
+ @TestTemplate
+ @Override
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ logger.warn("Caught exception while sending", e);
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
+ anonSender.send(message);
+ sentAnon.countDown();
+ } catch (Exception e) {
+ logger.warn("Caught exception while sending", e);
+ }
+ });
Review Comment:
These can change to just bare _Runnable_, given one intent of using an executor was to not use Thread at all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#issuecomment-2654822950
I brought this commit into https://github.com/apache/activemq-artemis/pull/5498 and made a few adjustments!
Thanks @gaohoward
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1726977714
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -1096,6 +1096,7 @@ public boolean checkMemory(boolean runOnFailure, Runnable runWhenAvailableParame
if (isFull()) {
if (runOnFailure && runWhenAvailable != null) {
addToBlockList(runWhenAvailable, blockedCallback);
+ pagingManager.addBlockedStore(this);
Review Comment:
me neither. Need @clebertsuconic's expertise. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1727224027
##########
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java:
##########
@@ -1096,6 +1096,7 @@ public boolean checkMemory(boolean runOnFailure, Runnable runWhenAvailableParame
if (isFull()) {
if (runOnFailure && runWhenAvailable != null) {
addToBlockList(runWhenAvailable, blockedCallback);
+ pagingManager.addBlockedStore(this);
Review Comment:
PR updated. The code path is really processing many conditions, not sure what's impact of that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1722970709
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
Review Comment:
This test class name seems rather generic for what its testing and how. It is testing handling of 'FAIL' policy behaviour while the disk is full/not, and reusing the GlobalDiskFullTest tests in part to do it. Something like GlobalDiskFullFailPolicyTest to link the two might be nicer.
EDIT: actually, it is overriding (without indicating so) the only test method in GlobalDiskFullTest, so it doesnt seem worth even extending that class in the current state. They are still testing related things however so naming them accordingly would still make sense.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ this.addAdditionalAcceptors(server);
+ this.configureAddressPolicy(server);
+ this.configureBrokerSecurity(server);
+ this.addConfiguration(server);
+ server.start();
+ this.createAddressAndQueues(server);
+ return server;
+ }
Review Comment:
Given that GlobalDiskFullTest manages not to override this method at all, and the ancestor parent AmqpTestSupport already sets many of these options to these same values, and offers specific methods for tweaking the config / address settings / etc to the tests needs, I'd expect something far simpler and cleaner than this can/should be done here.
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
Review Comment:
Should be a comment not javadoc, and be indented properly (e.g the URL). See other tests (e.g GlobalDiskFullTest this currently subclasses)
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
Review Comment:
The Jira indicates an issue with either policy of either FAIL or DROP...but the code change seems like it could only affect FAIL, because its behind a gate checking for FAIL? Is another test and possibly fix needed?
The test method code itself below for FAIL looks essentially identical to the existing GlobalDiskFullTest, which I presume is covering PAGE, so if the same works for DROP then perhaps they could all share the same overall base test, setting the specific mode covered via the constructor (either via ParameterizedTestExtension usage, or actual trivial subclasses of a base class)
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ this.addAdditionalAcceptors(server);
+ this.configureAddressPolicy(server);
+ this.configureBrokerSecurity(server);
+ this.addConfiguration(server);
+ server.start();
+ this.createAddressAndQueues(server);
+ return server;
+ }
+
+ @Test
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ threadWithName.start();
+
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
+ anonSender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ sentAnon.countDown();
+ }
+ });
+
+ threadWithAnon.start();
Review Comment:
Create an Executor to use, will be cleaner and the the test can also ensure it is e.g shutdownNow at the end by using the runAfter(..) mechanism (various other tests that do this).
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
Review Comment:
Will need to lose the star import, it will cause a checkstyle failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gemmellr (via GitHub)" <gi...@apache.org>.
gemmellr commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1731005986
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullTest.java:
##########
@@ -19,38 +19,99 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.lang.invoke.MethodHandles;
import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
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.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.artemis.utils.Wait;
import org.apache.activemq.transport.amqp.client.AmqpClient;
import org.apache.activemq.transport.amqp.client.AmqpConnection;
import org.apache.activemq.transport.amqp.client.AmqpMessage;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+@ExtendWith(ParameterizedTestExtension.class)
public class GlobalDiskFullTest extends AmqpClientTestSupport {
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameter(index = 0)
+ public AddressFullMessagePolicy addressFullPolicy;
+
+ @Parameters(name = "addressFullPolicy={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {AddressFullMessagePolicy.FAIL}, {AddressFullMessagePolicy.DROP}, {AddressFullMessagePolicy.PAGE}
+ });
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(addressFullPolicy);
+ server.getConfiguration().addAddressSetting(getQueueName(), addressSettings);
+ }
@Override
protected void addConfiguration(ActiveMQServer server) {
Configuration serverConfig = server.getConfiguration();
serverConfig.setDiskScanPeriod(100);
}
- @Test
+ @TestTemplate
public void testProducerOnDiskFull() throws Exception {
- FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
- final CountDownLatch latch = new CountDownLatch(1);
+
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor();
+
+ AtomicBoolean diskUsageOk = new AtomicBoolean(true);
+ AtomicInteger checkValid = new AtomicInteger(0);
+
monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
- latch.countDown();
+
+ if (checkValid.get() == -1) {
+ return;
+ }
+
+ checkValid.incrementAndGet();
+
+ double usage = FileStoreMonitor.calculateUsage(usableSpace, totalSpace);
+
+ if (type == FileStoreMonitor.FileStoreMonitorType.MaxDiskUsage && ok && usage < monitor.getMaxUsage()) {
+ diskUsageOk.set(true);
+ } else if (type == FileStoreMonitor.FileStoreMonitorType.MaxDiskUsage && !ok && usage >= monitor.getMaxUsage()) {
+ diskUsageOk.set(false);
+ } else {
+ logger.warn("invalid state, usableSpace: {}, totalSpace: {}, ok: {}, type: {}", usableSpace, totalSpace, ok, type);
+ checkValid.set(-1);
+ }
});
- assertTrue(latch.await(1, TimeUnit.MINUTES));
+ Wait.assertTrue(() -> checkValid.get() > 0, 1000);
+ Wait.assertTrue(() -> diskUsageOk.get(), 1000);
Review Comment:
Another benefit of switching back to simple latch inspection, most of these separate waits go and we dont burn any more time than strictly needed (i.e just the time the monitor callback takes to fire, no additional sleep intervals from the Wait usage on top).
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullTest.java:
##########
@@ -19,38 +19,99 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.lang.invoke.MethodHandles;
import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
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.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.artemis.utils.Wait;
import org.apache.activemq.transport.amqp.client.AmqpClient;
import org.apache.activemq.transport.amqp.client.AmqpConnection;
import org.apache.activemq.transport.amqp.client.AmqpMessage;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+@ExtendWith(ParameterizedTestExtension.class)
public class GlobalDiskFullTest extends AmqpClientTestSupport {
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Parameter(index = 0)
+ public AddressFullMessagePolicy addressFullPolicy;
+
+ @Parameters(name = "addressFullPolicy={0}")
+ public static Collection<Object[]> parameters() {
+ return Arrays.asList(new Object[][] {
+ {AddressFullMessagePolicy.FAIL}, {AddressFullMessagePolicy.DROP}, {AddressFullMessagePolicy.PAGE}
+ });
+ }
+
+ @Override
+ protected void configureAddressPolicy(ActiveMQServer server) {
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(addressFullPolicy);
+ server.getConfiguration().addAddressSetting(getQueueName(), addressSettings);
+ }
@Override
protected void addConfiguration(ActiveMQServer server) {
Configuration serverConfig = server.getConfiguration();
serverConfig.setDiskScanPeriod(100);
}
- @Test
+ @TestTemplate
public void testProducerOnDiskFull() throws Exception {
- FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
- final CountDownLatch latch = new CountDownLatch(1);
+
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor();
+
+ AtomicBoolean diskUsageOk = new AtomicBoolean(true);
+ AtomicInteger checkValid = new AtomicInteger(0);
+
monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
- latch.countDown();
+
+ if (checkValid.get() == -1) {
+ return;
+ }
+
+ checkValid.incrementAndGet();
+
+ double usage = FileStoreMonitor.calculateUsage(usableSpace, totalSpace);
+
+ if (type == FileStoreMonitor.FileStoreMonitorType.MaxDiskUsage && ok && usage < monitor.getMaxUsage()) {
+ diskUsageOk.set(true);
+ } else if (type == FileStoreMonitor.FileStoreMonitorType.MaxDiskUsage && !ok && usage >= monitor.getMaxUsage()) {
+ diskUsageOk.set(false);
+ } else {
+ logger.warn("invalid state, usableSpace: {}, totalSpace: {}, ok: {}, type: {}", usableSpace, totalSpace, ok, type);
+ checkValid.set(-1);
+ }
Review Comment:
This seems overly complex and a bit fragile in terms of the asserting, with its multiple independent values and the different threads in use probably making asserts span callback firings.
There can be multiple callbacks added. Instead of trying to make one complex callback work, it would probably be better and clearer to add specific trivial callbacks for each stage check at the appropriate point, which only trips its latch when the desired state is reached. You can then assert that the desired state has been reached (or not) while waiting on the latch after provoking the state change, but also if desired beforehand assert that it has not been reached yet by checking the latch doesn't trip yet. Then before the next stage, add a new callback to verify that new stage.
Eg. add one immediately (essentially the same as the original test (but its latch gated so it only trips if the state is usage-full), then use it to verify the not-blocked and blocked state before and after setting the maxUsage to 0. Once that is done add a new callback with its own latch, do the sends, inspect latch to verify state is still blocked (not tripped), then after setting the max to 100 verify that it becomes unblocked (tripped).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1723913937
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ this.addAdditionalAcceptors(server);
+ this.configureAddressPolicy(server);
+ this.configureBrokerSecurity(server);
+ this.addConfiguration(server);
+ server.start();
+ this.createAddressAndQueues(server);
+ return server;
+ }
+
+ @Test
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ threadWithName.start();
+
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
+ anonSender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
Review Comment:
logger please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#issuecomment-2654855116
actually, I sent a new PR for this:
https://github.com/apache/activemq-artemis/pull/5502
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "clebertsuconic (via GitHub)" <gi...@apache.org>.
clebertsuconic commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1723913937
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
+ server.getConfiguration().addAddressSetting("TEST",addressSettings);
+ this.addAdditionalAcceptors(server);
+ this.configureAddressPolicy(server);
+ this.configureBrokerSecurity(server);
+ this.addConfiguration(server);
+ server.start();
+ this.createAddressAndQueues(server);
+ return server;
+ }
+
+ @Test
+ public void testProducerOnDiskFull() throws Exception {
+ FileStoreMonitor monitor = ((ActiveMQServerImpl)server).getMonitor().setMaxUsage(0.0);
+ final CountDownLatch latch = new CountDownLatch(1);
+ monitor.addCallback((usableSpace, totalSpace, ok, type) -> {
+ latch.countDown();
+ });
+
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ AmqpClient client = createAmqpClient(new URI("tcp://localhost:" + AMQP_PORT));
+ AmqpConnection connection = addConnection(client.connect());
+
+ try {
+ AmqpSession session = connection.createSession();
+ AmqpSender sender = session.createSender("TEST");
+ byte[] payload = new byte[1000];
+
+ AmqpSender anonSender = session.createSender();
+
+ CountDownLatch sentWithName = new CountDownLatch(1);
+ CountDownLatch sentAnon = new CountDownLatch(1);
+
+ Thread threadWithName = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ sender.setSendTimeout(-1);
+ sender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ sentWithName.countDown();
+ }
+ });
+
+ threadWithName.start();
+
+
+ Thread threadWithAnon = new Thread(() -> {
+ try {
+ final AmqpMessage message = new AmqpMessage();
+ message.setBytes(payload);
+ anonSender.setSendTimeout(-1);
+ message.setAddress(getQueueName());
+ anonSender.send(message);
+ } catch (Exception e) {
+ e.printStackTrace();
Review Comment:
use logger.warn here please/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1724297983
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
Review Comment:
I'll add a DROP test and see how it behaves.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1724316247
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AMQPBlockingTest.java:
##########
@@ -0,0 +1,128 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.transport.amqp.client.*;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class AMQPBlockingTest extends GlobalDiskFullTest {
+
+ @Override
+ protected ActiveMQServer createServer(int port) throws Exception {
+ ActiveMQServer server = this.createServer(true, true);
+ server.getConfiguration().getAcceptorConfigurations().clear();
+ server.getConfiguration().getAcceptorConfigurations().add(this.addAcceptorConfiguration(server, port));
+ server.getConfiguration().setName("localhost");
+ server.getConfiguration().setJournalDirectory(server.getConfiguration().getJournalDirectory() + port);
+ server.getConfiguration().setBindingsDirectory(server.getConfiguration().getBindingsDirectory() + port);
+ server.getConfiguration().setPagingDirectory(server.getConfiguration().getPagingDirectory() + port);
+ server.getConfiguration().setMessageExpiryScanPeriod(5000L);
+ AddressSettings addressSettings = new AddressSettings();
+ addressSettings.setAddressFullMessagePolicy(AddressFullMessagePolicy.FAIL);
Review Comment:
actually the test pass with DROP. The fix only affects FAIL. Looks like the DROP was handled correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on code in PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#discussion_r1726975123
##########
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/GlobalDiskFullFailPolicyTest.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.integration.amqp;
+
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.files.FileStoreMonitor;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameter;
+import org.apache.activemq.artemis.tests.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.activemq.artemis.tests.extensions.parameterized.Parameters;
+import org.apache.activemq.transport.amqp.client.AmqpClient;
+import org.apache.activemq.transport.amqp.client.AmqpConnection;
+import org.apache.activemq.transport.amqp.client.AmqpMessage;
+import org.apache.activemq.transport.amqp.client.AmqpSender;
+import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.invoke.MethodHandles;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class GlobalDiskFullFailPolicyTest extends GlobalDiskFullTest {
Review Comment:
ah ok. will update. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact
Re: [PR] ARTEMIS-5002 AMQP producer not unblock if the disk space is freed [activemq-artemis]
Posted by "gaohoward (via GitHub)" <gi...@apache.org>.
gaohoward commented on PR #5160:
URL: https://github.com/apache/activemq-artemis/pull/5160#issuecomment-2297883660
@clebertsuconic can you take a look of this PR see if the fix makes sense? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For additional commands, e-mail: gitbox-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact