[jira] [Work logged] (ARTEMIS-4670) Slow performance with Core large messages and JDBC
[ https://issues.apache.org/jira/browse/ARTEMIS-4670?focusedWorklogId=908461=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908461 ] ASF GitHub Bot logged work on ARTEMIS-4670: --- Author: ASF GitHub Bot Created on: 06/Mar/24 02:59 Start Date: 06/Mar/24 02:59 Worklog Time Spent: 10m Work Description: clebertsuconic commented on PR #4842: URL: https://github.com/apache/activemq-artemis/pull/4842#issuecomment-1979990456 there are test failures.. converting to draft for now. Issue Time Tracking --- Worklog Id: (was: 908461) Time Spent: 20m (was: 10m) > Slow performance with Core large messages and JDBC > -- > > Key: ARTEMIS-4670 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4670 > Project: ActiveMQ Artemis > Issue Type: Bug >Reporter: Clebert Suconic >Priority: Major > Time Spent: 20m > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (ARTEMIS-4671) set the filter expression on the queues under the same address, queues receive a different number of messages
Ke Xu created ARTEMIS-4671:
--
Summary: set the filter expression on the queues under the same
address, queues receive a different number of messages
Key: ARTEMIS-4671
URL: https://issues.apache.org/jira/browse/ARTEMIS-4671
Project: ActiveMQ Artemis
Issue Type: Bug
Components: Broker
Affects Versions: 2.19.1
Environment: OS Version: 5.14.0-70.13.1.el9_0.x86_64
JDK Version:
java version "1.8.0_281"
Java(TM) SE Runtime Environment (build 1.8.0_281-b09)
Java HotSpot(TM) 64-Bit Server VM (build 25.281-b09, mixed mode)
Web Server: Apache Tomcat/9.0.35
Artemis: 2.19.1
Reporter: Ke Xu
Attachments: broker-211.xml, broker-215.xml,
image-2024-03-06-09-48-41-501.png
I am experiencing an issue with ActiveMQ Artemis. I created an address named
IN.ADDRESS with the address mode type set to "multiple". We also created two
queues (IN.ADDRESS.Q1 and IN.ADDRESS.Q2) and set the filter expression on the
queues using the CORE API. Regardless of whether the filter is the same or not,
for example, when we send 2000 messages to the address that do not meet the
filter, I am sure the messages are not being filtered. However, the queues
receive a different number of messages, which suggests that some messages may
be lost in the routing process.If I remove the filter expression on one queue,
the message count is correct.
Here is my code for create queue:
{code:java}
public Queue outFss1Queue() {
try {
String filter =
ConvertUtils.genFilter(staticProperties.getServiceCode(),
SsConstant.XML_MSG_TYPE_FSS1);
SimpleString str = new
SimpleString(staticProperties.getFss1InQueue());
ClientSession.QueueQuery query = session.queueQuery(str);
QueueConfiguration queueConf = new
QueueConfiguration(staticProperties.getFss1InQueue());
queueConf.setAddress(staticProperties.getFss1InTopic());
queueConf.setDurable(true);
queueConf.setAutoCreated(true);
queueConf.setAutoCreateAddress(true);
queueConf.setRoutingType(RoutingType.MULTICAST);
queueConf.setFilterString(filter);
if (!query.isExists()) {
queueConf.setEnabled(false);
session.createQueue(queueConf);
session.start();
} else {
ClientRequestor requestor = new ClientRequestor(session,
"activemq.management");
ClientMessage message = session.createMessage(false);
ManagementHelper.putOperationInvocation(message,
ResourceNames.BROKER, "updateQueue", queueConf.toJSON());
session.start();
ClientMessage response = requestor.request(message);
Object resResult = ManagementHelper.getResult(response);
requestor.close();
}
} catch (Exception e) {
log.error("Init fss1 queue failed, error:", e);
throw new RuntimeException("Init fss1 queue failed");
}
return new ActiveMQQueue(staticProperties.getFss1InQueue());
} {code}
Here is my filter expression on the queue. In this case, my queues have the
same filter expression.
{code:java}
XPATH /IMFRoot/Data/PrimaryKey/FlightKey/FlightDirection[text() =
D or @OldValue =
D]|/IMFRoot/SysInfo/OperationMode[text()=DEL][/IMFRoot/Data/PrimaryKey/FlightKey/FlightDirection[text()
= D or @OldValue = D]] {code}
Here is my broker.xml configuration file. We have two servers in the Artemis
cluster, so we have two broker.xml files. I have upload my configuration as an
attachment.
---broker-211.xml---
{code:java}
http://www.w3.org/2001/XMLSchema-instance;
xmlns:xi="http://www.w3.org/2001/XInclude;
xsi:schemaLocation="urn:activemq
/schema/artemis-configuration.xsd"> http://www.w3.org/2001/XMLSchema-instance;
xsi:schemaLocation="urn:activemq:core "> slave
true
80
ASYNCIO
data/paging
data/bindings
data/journal
data/large-messages
true
2
10
4096
10M
5
4096
tcp://192.168.201.211:61616
tcp://192.168.201.215:61616
5000
90
true
12
6
HALT
38
tcp://192.168.201.211:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true;supportAdvisory=false;suppressInternalManagementObjects=false
[jira] [Work logged] (ARTEMIS-4670) Slow performance with Core large messages and JDBC
[ https://issues.apache.org/jira/browse/ARTEMIS-4670?focusedWorklogId=908447=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908447 ] ASF GitHub Bot logged work on ARTEMIS-4670: --- Author: ASF GitHub Bot Created on: 06/Mar/24 00:11 Start Date: 06/Mar/24 00:11 Worklog Time Spent: 10m Work Description: clebertsuconic opened a new pull request, #4842: URL: https://github.com/apache/activemq-artemis/pull/4842 (no comment) Issue Time Tracking --- Worklog Id: (was: 908447) Remaining Estimate: 0h Time Spent: 10m > Slow performance with Core large messages and JDBC > -- > > Key: ARTEMIS-4670 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4670 > Project: ActiveMQ Artemis > Issue Type: Bug >Reporter: Clebert Suconic >Priority: Major > Time Spent: 10m > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (ARTEMIS-4670) Slow performance with Core large messages and JDBC
[ https://issues.apache.org/jira/browse/ARTEMIS-4670?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Clebert Suconic updated ARTEMIS-4670: - Summary: Slow performance with Core large messages and JDBC (was: Slow performance with AMQP large messages and JDBC) > Slow performance with Core large messages and JDBC > -- > > Key: ARTEMIS-4670 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4670 > Project: ActiveMQ Artemis > Issue Type: Bug >Reporter: Clebert Suconic >Priority: Major > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (ARTEMIS-4670) Slow performance with AMQP large messages and JDBC
[ https://issues.apache.org/jira/browse/ARTEMIS-4670?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17823808#comment-17823808 ] Clebert Suconic commented on ARTEMIS-4670: -- there is blocking wait when using large messages and JDBC: https://github.com/rh-messaging/activemq-artemis/blob/28df0dea98f4ef4ddb517b3bb81891848d5d8e85/artemis-jdbc-store/src/main/java/org/apache/activemq/artemis/jdbc/store/file/JDBCSequentialFile.java#L337 which won't scale. > Slow performance with AMQP large messages and JDBC > -- > > Key: ARTEMIS-4670 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4670 > Project: ActiveMQ Artemis > Issue Type: Bug >Reporter: Clebert Suconic >Priority: Major > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (ARTEMIS-4670) Slow performance with AMQP large messages and JDBC
Clebert Suconic created ARTEMIS-4670: Summary: Slow performance with AMQP large messages and JDBC Key: ARTEMIS-4670 URL: https://issues.apache.org/jira/browse/ARTEMIS-4670 Project: ActiveMQ Artemis Issue Type: Bug Reporter: Clebert Suconic -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[ https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908433=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908433 ] ASF GitHub Bot logged work on ARTEMIS-4668: --- Author: ASF GitHub Bot Created on: 05/Mar/24 22:29 Start Date: 05/Mar/24 22:29 Worklog Time Spent: 10m Work Description: clebertsuconic commented on PR #4840: URL: https://github.com/apache/activemq-artemis/pull/4840#issuecomment-1979744561 @tabish121 , @gemmellr I tried simplifying the changes as much as possible here... Issue Time Tracking --- Worklog Id: (was: 908433) Time Spent: 6h (was: 5h 50m) > Move AMQP Large Message File Handling away from Netty thread > > > Key: ARTEMIS-4668 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4668 > Project: ActiveMQ Artemis > Issue Type: Bug >Affects Versions: 2.32.0 >Reporter: Clebert Suconic >Assignee: Clebert Suconic >Priority: Major > Fix For: 2.33.0 > > Time Spent: 6h > Remaining Estimate: 0h > > Operations like file.open, file.close, and file.sync should be moved away > from the Netty Thread for AMQP Large Messages > This task now is about moving the processing for AMQP Messages. we may in a > near future also improve tunneled large messages. For now we will do for AMQP > messages only. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (ARTEMIS-4669) Clarify Large Messages around StorageManager usage
[ https://issues.apache.org/jira/browse/ARTEMIS-4669?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17823790#comment-17823790 ] ASF subversion and git services commented on ARTEMIS-4669: -- Commit 5ce70f9e37ba1bb7c56372c2e79f9e9d8c9d1fa5 in activemq-artemis's branch refs/heads/main from Clebert Suconic [ https://gitbox.apache.org/repos/asf?p=activemq-artemis.git;h=5ce70f9e37 ] ARTEMIS-4669 Clarify Storage Manager usage around large messages > Clarify Large Messages around StorageManager usage > -- > > Key: ARTEMIS-4669 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4669 > Project: ActiveMQ Artemis > Issue Type: Task >Reporter: Clebert Suconic >Priority: Major > Time Spent: 0.5h > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4669) Clarify Large Messages around StorageManager usage
[ https://issues.apache.org/jira/browse/ARTEMIS-4669?focusedWorklogId=908432=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908432 ] ASF GitHub Bot logged work on ARTEMIS-4669: --- Author: ASF GitHub Bot Created on: 05/Mar/24 22:27 Start Date: 05/Mar/24 22:27 Worklog Time Spent: 10m Work Description: clebertsuconic merged PR #4841: URL: https://github.com/apache/activemq-artemis/pull/4841 Issue Time Tracking --- Worklog Id: (was: 908432) Time Spent: 0.5h (was: 20m) > Clarify Large Messages around StorageManager usage > -- > > Key: ARTEMIS-4669 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4669 > Project: ActiveMQ Artemis > Issue Type: Task >Reporter: Clebert Suconic >Priority: Major > Time Spent: 0.5h > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4669) Clarify Large Messages around StorageManager usage
[ https://issues.apache.org/jira/browse/ARTEMIS-4669?focusedWorklogId=908431=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908431 ] ASF GitHub Bot logged work on ARTEMIS-4669: --- Author: ASF GitHub Bot Created on: 05/Mar/24 22:27 Start Date: 05/Mar/24 22:27 Worklog Time Spent: 10m Work Description: clebertsuconic commented on PR #4841: URL: https://github.com/apache/activemq-artemis/pull/4841#issuecomment-1979742208 this is a simple renaming of methods... if anyone has any concerns about these names I used.. please send a new commit... all I cared here is to not have createLargeMessage and largeMessageCreated on the API (which is something I did in the past.. but it was way confusing for me :) ) I am merging as my next PR is based no this. Issue Time Tracking --- Worklog Id: (was: 908431) Time Spent: 20m (was: 10m) > Clarify Large Messages around StorageManager usage > -- > > Key: ARTEMIS-4669 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4669 > Project: ActiveMQ Artemis > Issue Type: Task >Reporter: Clebert Suconic >Priority: Major > Time Spent: 20m > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4669) Clarify Large Messages around StorageManager usage
[ https://issues.apache.org/jira/browse/ARTEMIS-4669?focusedWorklogId=908424=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908424 ] ASF GitHub Bot logged work on ARTEMIS-4669: --- Author: ASF GitHub Bot Created on: 05/Mar/24 21:16 Start Date: 05/Mar/24 21:16 Worklog Time Spent: 10m Work Description: clebertsuconic opened a new pull request, #4841: URL: https://github.com/apache/activemq-artemis/pull/4841 (no comment) Issue Time Tracking --- Worklog Id: (was: 908424) Remaining Estimate: 0h Time Spent: 10m > Clarify Large Messages around StorageManager usage > -- > > Key: ARTEMIS-4669 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4669 > Project: ActiveMQ Artemis > Issue Type: Task >Reporter: Clebert Suconic >Priority: Major > Time Spent: 10m > Remaining Estimate: 0h > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (ARTEMIS-4669) Clarify Large Messages around StorageManager usage
[ https://issues.apache.org/jira/browse/ARTEMIS-4669?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17823776#comment-17823776 ] Clebert Suconic commented on ARTEMIS-4669: -- there are two methods in StorageManager that are confusing: createLargeMessage and largeMessageCreated one is to create a core large message file and object, the other is a callback for when large messages are created (including other protocols, currently only AMQP). These two methods should be renamed. Also validateFile is no longer needed since we stopped storing pending records on the journal. At some point we needed to make sure the file existed before we stored records in the storage for replication. Which is no longer the case. > Clarify Large Messages around StorageManager usage > -- > > Key: ARTEMIS-4669 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4669 > Project: ActiveMQ Artemis > Issue Type: Task >Reporter: Clebert Suconic >Priority: Major > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (ARTEMIS-4669) Clarify Large Messages around StorageManager usage
Clebert Suconic created ARTEMIS-4669: Summary: Clarify Large Messages around StorageManager usage Key: ARTEMIS-4669 URL: https://issues.apache.org/jira/browse/ARTEMIS-4669 Project: ActiveMQ Artemis Issue Type: Task Reporter: Clebert Suconic -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (AMQ-9394) Tech Preview: Virtual Thread support
[ https://issues.apache.org/jira/browse/AMQ-9394?focusedWorklogId=908422=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908422 ] ASF GitHub Bot logged work on AMQ-9394: --- Author: ASF GitHub Bot Created on: 05/Mar/24 20:55 Start Date: 05/Mar/24 20:55 Worklog Time Spent: 10m Work Description: mattrpav commented on PR #1121: URL: https://github.com/apache/activemq/pull/1121#issuecomment-1979620367 Newer PR here: #1172 Issue Time Tracking --- Worklog Id: (was: 908422) Time Spent: 50m (was: 40m) > Tech Preview: Virtual Thread support > > > Key: AMQ-9394 > URL: https://issues.apache.org/jira/browse/AMQ-9394 > Project: ActiveMQ Classic > Issue Type: New Feature >Reporter: Matt Pavlovich >Assignee: Matt Pavlovich >Priority: Major > Labels: #virtualthread > Fix For: 6.2.0 > > Time Spent: 50m > Remaining Estimate: 0h > > Initial technical preview of Virtual Thread support -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (AMQ-9394) Tech Preview: Virtual Thread support
[ https://issues.apache.org/jira/browse/AMQ-9394?focusedWorklogId=908423=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908423 ] ASF GitHub Bot logged work on AMQ-9394: --- Author: ASF GitHub Bot Created on: 05/Mar/24 20:55 Start Date: 05/Mar/24 20:55 Worklog Time Spent: 10m Work Description: mattrpav closed pull request #1121: WIP: [AMQ-9394] Tech Preview: Virtual Thread support URL: https://github.com/apache/activemq/pull/1121 Issue Time Tracking --- Worklog Id: (was: 908423) Time Spent: 1h (was: 50m) > Tech Preview: Virtual Thread support > > > Key: AMQ-9394 > URL: https://issues.apache.org/jira/browse/AMQ-9394 > Project: ActiveMQ Classic > Issue Type: New Feature >Reporter: Matt Pavlovich >Assignee: Matt Pavlovich >Priority: Major > Labels: #virtualthread > Fix For: 6.2.0 > > Time Spent: 1h > Remaining Estimate: 0h > > Initial technical preview of Virtual Thread support -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (AMQ-9394) Tech Preview: Virtual Thread support
[ https://issues.apache.org/jira/browse/AMQ-9394?focusedWorklogId=908421=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908421 ] ASF GitHub Bot logged work on AMQ-9394: --- Author: ASF GitHub Bot Created on: 05/Mar/24 20:54 Start Date: 05/Mar/24 20:54 Worklog Time Spent: 10m Work Description: mattrpav opened a new pull request, #1172: URL: https://github.com/apache/activemq/pull/1172 (no comment) Issue Time Tracking --- Worklog Id: (was: 908421) Time Spent: 40m (was: 0.5h) > Tech Preview: Virtual Thread support > > > Key: AMQ-9394 > URL: https://issues.apache.org/jira/browse/AMQ-9394 > Project: ActiveMQ Classic > Issue Type: New Feature >Reporter: Matt Pavlovich >Assignee: Matt Pavlovich >Priority: Major > Labels: #virtualthread > Fix For: 6.2.0 > > Time Spent: 40m > Remaining Estimate: 0h > > Initial technical preview of Virtual Thread support -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work started] (AMQ-9394) Tech Preview: Virtual Thread support
[ https://issues.apache.org/jira/browse/AMQ-9394?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Work on AMQ-9394 started by Matt Pavlovich. --- > Tech Preview: Virtual Thread support > > > Key: AMQ-9394 > URL: https://issues.apache.org/jira/browse/AMQ-9394 > Project: ActiveMQ Classic > Issue Type: New Feature >Reporter: Matt Pavlovich >Assignee: Matt Pavlovich >Priority: Major > Labels: #virtualthread > Fix For: 6.2.0 > > Time Spent: 0.5h > Remaining Estimate: 0h > > Initial technical preview of Virtual Thread support -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[ https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908399=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908399 ] ASF GitHub Bot logged work on ARTEMIS-4668: --- Author: ASF GitHub Bot Created on: 05/Mar/24 19:06 Start Date: 05/Mar/24 19:06 Worklog Time Spent: 10m Work Description: clebertsuconic commented on PR #4840: URL: https://github.com/apache/activemq-artemis/pull/4840#issuecomment-1979453938 I'm thinking on reverting some of the readBytes(Delivery) methods and only call the onMessageComplete for the Large Message case. will be working on it now. Issue Time Tracking --- Worklog Id: (was: 908399) Time Spent: 5h 50m (was: 5h 40m) > Move AMQP Large Message File Handling away from Netty thread > > > Key: ARTEMIS-4668 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4668 > Project: ActiveMQ Artemis > Issue Type: Bug >Affects Versions: 2.32.0 >Reporter: Clebert Suconic >Assignee: Clebert Suconic >Priority: Major > Fix For: 2.33.0 > > Time Spent: 5h 50m > Remaining Estimate: 0h > > Operations like file.open, file.close, and file.sync should be moved away > from the Netty Thread for AMQP Large Messages > This task now is about moving the processing for AMQP Messages. we may in a > near future also improve tunneled large messages. For now we will do for AMQP > messages only. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[ https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908398=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908398 ] ASF GitHub Bot logged work on ARTEMIS-4668: --- Author: ASF GitHub Bot Created on: 05/Mar/24 18:50 Start Date: 05/Mar/24 18:50 Worklog Time Spent: 10m Work Description: clebertsuconic commented on PR #4840: URL: https://github.com/apache/activemq-artemis/pull/4840#issuecomment-1979429610 @gemmellr , @tabish121 I added 3 temporary commits to this PR. Let me know if you have any questions and I will squash them before setting it ready to merge. Issue Time Tracking --- Worklog Id: (was: 908398) Time Spent: 5h 40m (was: 5.5h) > Move AMQP Large Message File Handling away from Netty thread > > > Key: ARTEMIS-4668 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4668 > Project: ActiveMQ Artemis > Issue Type: Bug >Affects Versions: 2.32.0 >Reporter: Clebert Suconic >Assignee: Clebert Suconic >Priority: Major > Fix For: 2.33.0 > > Time Spent: 5h 40m > Remaining Estimate: 0h > > Operations like file.open, file.close, and file.sync should be moved away > from the Netty Thread for AMQP Large Messages > This task now is about moving the processing for AMQP Messages. we may in a > near future also improve tunneled large messages. For now we will do for AMQP > messages only. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908397=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908397
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:46
Start Date: 05/Mar/24 18:46
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513335419
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageWriter.java:
##
@@ -144,21 +159,28 @@ private void resume() {
private void tryDelivering() {
// This is discounting some bytes due to Transfer payload
- final int frameSize =
protonSender.getSession().getConnection().getTransport().getOutboundFrameSizeLimit()
- 50 - (delivery.getTag() != null ? delivery.getTag().length : 0);
try {
+ if (this.delivery == null) {
+this.delivery = serverSender.createDelivery(reference, (int)
this.message.getMessageFormat());
+if (sessionSPI.invokeOutgoing(message,
(ActiveMQProtonRemotingConnection)
sessionSPI.getTransportConnection().getProtocolConnection()) != null) {
+ return;
+}
Review Comment:
commit to be squashed 55ce243a690a1dcae26587ea6e962effb896f039 is addressing
this
Issue Time Tracking
---
Worklog Id: (was: 908397)
Time Spent: 5.5h (was: 5h 20m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 5.5h
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908391=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908391
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:15
Start Date: 05/Mar/24 18:15
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513290548
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageWriter.java:
##
@@ -144,21 +159,28 @@ private void resume() {
private void tryDelivering() {
// This is discounting some bytes due to Transfer payload
- final int frameSize =
protonSender.getSession().getConnection().getTransport().getOutboundFrameSizeLimit()
- 50 - (delivery.getTag() != null ? delivery.getTag().length : 0);
try {
+ if (this.delivery == null) {
+this.delivery = serverSender.createDelivery(reference, (int)
this.message.getMessageFormat());
+if (sessionSPI.invokeOutgoing(message,
(ActiveMQProtonRemotingConnection)
sessionSPI.getTransportConnection().getProtocolConnection()) != null) {
+ return;
+}
Review Comment:
will work on it..
Issue Time Tracking
---
Worklog Id: (was: 908391)
Time Spent: 5h 10m (was: 5h)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 5h 10m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908392=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908392
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:15
Start Date: 05/Mar/24 18:15
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513291190
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPTunneledCoreLargeMessageReader.java:
##
@@ -209,13 +208,15 @@ public Message readBytes(Delivery delivery) throws
Exception {
state = State.DONE;
-return result;
- }
+serverReceiver.onMessageComplete(delivery, result,
deliveryAnnotations);
- return null;
+close();
Review Comment:
to be squashed commit handling it.
Issue Time Tracking
---
Worklog Id: (was: 908392)
Time Spent: 5h 20m (was: 5h 10m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 5h 20m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908390=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908390
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:14
Start Date: 05/Mar/24 18:14
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513289227
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
}
- currentMessage.addBytes(dataBuffer);
+ serverReceiver.getConnection().disableAutoRead();
Review Comment:
between the disableAutoRead and the call to the executor nothing else would
throw an exception.
I added handling on the exception catch to enable it back, also I'm enabling
it back in the onException method I added.
Issue Time Tracking
---
Worklog Id: (was: 908390)
Time Spent: 5h (was: 4h 50m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 5h
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908387=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908387
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:13
Start Date: 05/Mar/24 18:13
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513285024
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java:
##
@@ -378,6 +389,16 @@ public AMQPConnectionCallback getConnectionCallback() {
return connectionCallback;
}
+ public void exception(Throwable e) {
+ logger.warn(e.getMessage(), e);
+ ErrorCondition error = new ErrorCondition();
+ error.setCondition(AmqpError.INTERNAL_ERROR);
+ error.setDescription("Unrecoverable error: " + (e.getMessage() == null ?
e.getClass().getSimpleName() : e.getMessage()));
+ getHandler().getConnection().setCondition(error);
+ getHandler().getConnection().close();
+ flush();
+ }
Review Comment:
I added a commit to be squashed addressing this
Issue Time Tracking
---
Worklog Id: (was: 908387)
Time Spent: 4h 40m (was: 4.5h)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 4h 40m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908388=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908388
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:13
Start Date: 05/Mar/24 18:13
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513286921
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java:
##
@@ -94,6 +94,17 @@ public class AMQPConnectionContext extends
ProtonInitializable implements EventH
private static final Logger logger =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ public void disableAutoRead() {
+ connectionCallback.getTransportConnection().setAutoRead(false);
+ handler.setReadable(false);
+ }
+
+ public void enableAutoRead() {
+ connectionCallback.getTransportConnection().setAutoRead(true);
+ getHandler().setReadable(true);
Review Comment:
enableAutoRead is now called from the Netty Thread with a runNow call
this was going to call flush which would transfer the execution anyways.
my commit to be squashed is addressing this
Issue Time Tracking
---
Worklog Id: (was: 908388)
Time Spent: 4h 50m (was: 4h 40m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 4h 50m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908385=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908385
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:12
Start Date: 05/Mar/24 18:12
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513284233
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
}
- currentMessage.addBytes(dataBuffer);
+ serverReceiver.getConnection().disableAutoRead();
- final AMQPLargeMessage result;
+ byte[] bytes = new byte[dataBuffer.remaining()];
+ dataBuffer.get(bytes);
- if (!delivery.isPartial()) {
-
currentMessage.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
- result = currentMessage;
- // We don't want a close to delete the file now, we've released the
resources.
- currentMessage = null;
- deliveryAnnotations = result.getDeliveryAnnotations();
- } else {
- result = null;
+ boolean partial = delivery.isPartial();
+
+ sessionSPI.execute(() -> addBytes(delivery, bytes, partial));
+
+ return partial;
+ }
+
+ private void validateFile(AMQPLargeMessage message) {
+ try {
+ message.validateFile();
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ close();
+ serverReceiver.connection.exception(e);
}
+ }
- return result;
+ private void addBytes(Delivery delivery, byte[] bytes, boolean isPartial) {
+ ReadableBuffer dataBuffer = ReadableBuffer.ByteBufferReader.wrap(bytes);
+ try {
+ logger.trace("Adding {} bytes on currentMessage={}, this={}",
dataBuffer.remaining(), currentMessage, this);
+ currentMessage.addBytes(dataBuffer);
+
+ if (!isPartial) {
+final AMQPLargeMessage message = currentMessage;
+
message.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
+logger.trace("finishing {} on {}", currentMessage, this);
+// We don't want a close to delete the file now, we've released
the resources.
+currentMessage = null;
+close();
Review Comment:
I added a commit to be squashed addressing this
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id =
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908386=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908386
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 18:12
Start Date: 05/Mar/24 18:12
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513284534
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPTunneledCoreMessageReader.java:
##
@@ -155,6 +149,10 @@ public Message readBytes(Delivery delivery) {
coreMessage.reloadPersistence(buffer,
sessionSPI.getCoreMessageObjectPools());
coreMessage.setMessageID(sessionSPI.getStorageManager().generateID());
- return coreMessage;
+ serverReceiver.onMessageComplete(delivery, coreMessage,
deliveryAnnotations);
+
+ close();
Review Comment:
I added a commit to be squashed addressing this
##
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java:
##
@@ -126,6 +126,9 @@ public final Channel getNettyChannel() {
@Override
public final void setAutoRead(boolean autoRead) {
channel.config().setAutoRead(autoRead);
+ if (autoRead) {
+ channel.read();
+ }
Review Comment:
I added a commit to be squashed addressing this
Issue Time Tracking
---
Worklog Id: (was: 908386)
Time Spent: 4.5h (was: 4h 20m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 4.5h
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908382=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908382
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 17:57
Start Date: 05/Mar/24 17:57
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513265169
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
}
- currentMessage.addBytes(dataBuffer);
+ serverReceiver.getConnection().disableAutoRead();
- final AMQPLargeMessage result;
+ byte[] bytes = new byte[dataBuffer.remaining()];
+ dataBuffer.get(bytes);
- if (!delivery.isPartial()) {
-
currentMessage.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
- result = currentMessage;
- // We don't want a close to delete the file now, we've released the
resources.
- currentMessage = null;
- deliveryAnnotations = result.getDeliveryAnnotations();
- } else {
- result = null;
+ boolean partial = delivery.isPartial();
+
+ sessionSPI.execute(() -> addBytes(delivery, bytes, partial));
+
+ return partial;
Review Comment:
the handler.readable is trying to handle that.. yes..
I'm calling the enable from the Netty thread on my next push.
I'm working on it now.
Issue Time Tracking
---
Worklog Id: (was: 908382)
Time Spent: 4h 10m (was: 4h)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 4h 10m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908372=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908372
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 17:17
Start Date: 05/Mar/24 17:17
Worklog Time Spent: 10m
Work Description: tabish121 commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513206030
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageWriter.java:
##
@@ -81,33 +84,54 @@ public boolean isWriting() {
public void close() {
if (!closed) {
try {
+try {
+ if (largeBodyReader != null) {
+ largeBodyReader.close();
+ }
+} catch (Exception e) {
+ // if we get an error only at this point, there's nothing else
we could do other than log.warn
+ logger.warn("{}", e.getMessage(), e);
+}
if (message != null) {
message.usageDown();
}
} finally {
-reset(true);
+resetClosed();
}
}
}
@Override
- public AMQPLargeMessageWriter open() {
- if (!closed) {
- throw new IllegalStateException("Trying to open an AMQP Large Message
writer that was not closed");
+ public AMQPLargeMessageWriter open(MessageReference reference) {
+ this.reference = reference;
+ this.message = (AMQPLargeMessage) reference.getMessage();
Review Comment:
This is storing the message which means a call to close() will call
usageDown on the reference but the way the code is now structured there isn't a
100% certainly that usageUp will always be called on the reference. It needs
to be called here likely so that a close will indeed not accidentally lower
usage when it hadn't been increased.
Issue Time Tracking
---
Worklog Id: (was: 908372)
Time Spent: 4h (was: 3h 50m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 4h
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908371=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908371
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 17:13
Start Date: 05/Mar/24 17:13
Worklog Time Spent: 10m
Work Description: gemmellr commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513187210
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPTunneledCoreLargeMessageReader.java:
##
@@ -209,13 +208,15 @@ public Message readBytes(Delivery delivery) throws
Exception {
state = State.DONE;
-return result;
- }
+serverReceiver.onMessageComplete(delivery, result,
deliveryAnnotations);
- return null;
+close();
Review Comment:
Doesnt feel like a reader should be closing itself. Was meant to be managed
by the interested party using it.
Issue Time Tracking
---
Worklog Id: (was: 908371)
Time Spent: 3h 50m (was: 3h 40m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 3h 50m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908369=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908369
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 17:10
Start Date: 05/Mar/24 17:10
Worklog Time Spent: 10m
Work Description: gemmellr commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513095440
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java:
##
@@ -378,6 +389,16 @@ public AMQPConnectionCallback getConnectionCallback() {
return connectionCallback;
}
+ public void exception(Throwable e) {
+ logger.warn(e.getMessage(), e);
+ ErrorCondition error = new ErrorCondition();
+ error.setCondition(AmqpError.INTERNAL_ERROR);
+ error.setDescription("Unrecoverable error: " + (e.getMessage() == null ?
e.getClass().getSimpleName() : e.getMessage()));
+ getHandler().getConnection().setCondition(error);
+ getHandler().getConnection().close();
+ flush();
+ }
Review Comment:
This is being called from the 'session executor' so it isnt thread safe in
its use of the connection, which should only be used on the connection thread.
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageWriter.java:
##
@@ -144,21 +159,28 @@ private void resume() {
private void tryDelivering() {
// This is discounting some bytes due to Transfer payload
- final int frameSize =
protonSender.getSession().getConnection().getTransport().getOutboundFrameSizeLimit()
- 50 - (delivery.getTag() != null ? delivery.getTag().length : 0);
try {
+ if (this.delivery == null) {
+this.delivery = serverSender.createDelivery(reference, (int)
this.message.getMessageFormat());
+if (sessionSPI.invokeOutgoing(message,
(ActiveMQProtonRemotingConnection)
sessionSPI.getTransportConnection().getProtocolConnection()) != null) {
+ return;
+}
Review Comment:
This looks like it is creating a delivery, then potentially immediately
deciding not to send it. Should that not be the other way around? EDIT: Like it
was originally when the code lived in writeBytes.
##
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/remoting/impl/netty/NettyConnection.java:
##
@@ -126,6 +126,9 @@ public final Channel getNettyChannel() {
@Override
public final void setAutoRead(boolean autoRead) {
channel.config().setAutoRead(autoRead);
+ if (autoRead) {
+ channel.read();
+ }
Review Comment:
I believe Netty does this itself when setting autoRead true, if it wasnt
before. This here does it _every_ time, rather than just if it changed to true.
Since I think its also possible it actually reads there and then (if on the
correct thread), probably dont want to do it twice or more.
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
}
- currentMessage.addBytes(dataBuffer);
+ serverReceiver.getConnection().disableAutoRead();
Review Comment:
If it fails after this point, but before its enabled in the other thread (so
earlier comment about safety seem to
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908364=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908364
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 16:24
Start Date: 05/Mar/24 16:24
Worklog Time Spent: 10m
Work Description: tabish121 commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513124572
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
}
- currentMessage.addBytes(dataBuffer);
+ serverReceiver.getConnection().disableAutoRead();
- final AMQPLargeMessage result;
+ byte[] bytes = new byte[dataBuffer.remaining()];
+ dataBuffer.get(bytes);
- if (!delivery.isPartial()) {
-
currentMessage.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
- result = currentMessage;
- // We don't want a close to delete the file now, we've released the
resources.
- currentMessage = null;
- deliveryAnnotations = result.getDeliveryAnnotations();
- } else {
- result = null;
+ boolean partial = delivery.isPartial();
+
+ sessionSPI.execute(() -> addBytes(delivery, bytes, partial));
+
+ return partial;
+ }
+
+ private void validateFile(AMQPLargeMessage message) {
+ try {
+ message.validateFile();
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ close();
+ serverReceiver.connection.exception(e);
}
+ }
- return result;
+ private void addBytes(Delivery delivery, byte[] bytes, boolean isPartial) {
+ ReadableBuffer dataBuffer = ReadableBuffer.ByteBufferReader.wrap(bytes);
+ try {
+ logger.trace("Adding {} bytes on currentMessage={}, this={}",
dataBuffer.remaining(), currentMessage, this);
+ currentMessage.addBytes(dataBuffer);
+
+ if (!isPartial) {
+final AMQPLargeMessage message = currentMessage;
+
message.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
+logger.trace("finishing {} on {}", currentMessage, this);
+// We don't want a close to delete the file now, we've released
the resources.
+currentMessage = null;
+close();
+serverReceiver.connection.runNow(() ->
serverReceiver.onMessageComplete(delivery, message,
message.getDeliveryAnnotations()));
+ }
+ } catch (Throwable e) {
+ logger.warn(e.getMessage(), e);
+ close();
+ serverReceiver.connection.exception(e);
Review Comment:
The addBytes is not running on the connection thread so this call will end
up operating on the proton bits outside the connection thread leading to
possible inconsistent state when handling this.
Issue Time Tracking
---
Worklog Id: (was: 908364)
Time Spent: 3.5h (was: 3h 20m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>
[jira] [Commented] (ARTEMIS-4666) Federated queue consumers do not receive messages on classic clients
[ https://issues.apache.org/jira/browse/ARTEMIS-4666?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17823693#comment-17823693 ] Timothy A. Bish commented on ARTEMIS-4666: -- Likely then there is some interplay in core federation and openwire clients that is not accounted for. I would suggest having a look at the new AMQP federation over [AMQP broker connections|https://activemq.apache.org/components/artemis/documentation/latest/amqp-broker-connections.html#federation] although you might need to wait for the next release for the latest round of fixes there as well. The other option is to create an integration test that demonstrates the issue to try and help getting someone to look into it. > Federated queue consumers do not receive messages on classic clients > > > Key: ARTEMIS-4666 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4666 > Project: ActiveMQ Artemis > Issue Type: Bug > Components: Federation >Affects Versions: 2.32.0 >Reporter: Josh Byster >Priority: Major > > Federated queues with A upstream and downstream to B do not seem to work as > expected when the client JMS implementation is ActiveMQ Classic v5.16.2 (used > in my example, but also verified the issue is present with v5.18.3 and > v6.0.1). With Artemis JMS as the client, it seems to work as expected. > When running a producer on A and a consumer on B with the classic > org.apache.activemq.ActiveMQConnectionFactory, the consumer on B does not > consume any messages that the producer sends. When B is restarted, it then > consumes the messages. > This works properly with ActiveMQ Artemis > org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory. I've > created a minimal reproducible example > [here|https://github.com/josh-byster/artemis-classic-consumer-bug/tree/master]. > Running server1 and server2 and then first starting up the Consumer then > running the Producer class, we can see that no messages are logged in console > by the consumer. When you restart the consumer, the messages are consumed. > This is an issue whether you attach a MessageListener or you call receive() > directly. If you switch the ActiveMQConnectionFactory implementation to > Artemis, it works as expected. > I don't think this necessarily warrants a fix if it's an issue specifically > with the classic client, since the solution is just to upgrade clients to > Artemis. However, if it is something that can be patched on the server, that > would be great. I do, however, think it would be good to note this down in > the docs that it's not supported with classic clients, since I spent a while > debugging it. However, most other features do work as expected with the > clients running the classic version, which is much appreciated as it makes > migration significantly easier. > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908361=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908361
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 15:59
Start Date: 05/Mar/24 15:59
Worklog Time Spent: 10m
Work Description: tabish121 commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513082939
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPTunneledCoreMessageReader.java:
##
@@ -155,6 +149,10 @@ public Message readBytes(Delivery delivery) {
coreMessage.reloadPersistence(buffer,
sessionSPI.getCoreMessageObjectPools());
coreMessage.setMessageID(sessionSPI.getStorageManager().generateID());
- return coreMessage;
+ serverReceiver.onMessageComplete(delivery, coreMessage,
deliveryAnnotations);
+
+ close();
Review Comment:
I'm saying its wrong and I'm saying its thread unsafe. And if I need to I'm
saying -1 until I'm sure it is
Issue Time Tracking
---
Worklog Id: (was: 908361)
Time Spent: 3h 20m (was: 3h 10m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 3h 20m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908360=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908360
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 15:55
Start Date: 05/Mar/24 15:55
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513076021
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPTunneledCoreMessageReader.java:
##
@@ -155,6 +149,10 @@ public Message readBytes(Delivery delivery) {
coreMessage.reloadPersistence(buffer,
sessionSPI.getCoreMessageObjectPools());
coreMessage.setMessageID(sessionSPI.getStorageManager().generateID());
- return coreMessage;
+ serverReceiver.onMessageComplete(delivery, coreMessage,
deliveryAnnotations);
+
+ close();
Review Comment:
I changed the contract there. so the thing is closing itself before calling
the onComplete.
I was going to use that to release the resources.. but it's not needed any
more.
Issue Time Tracking
---
Worklog Id: (was: 908360)
Time Spent: 3h 10m (was: 3h)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (ARTEMIS-4666) Federated queue consumers do not receive messages on classic clients
[ https://issues.apache.org/jira/browse/ARTEMIS-4666?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17823688#comment-17823688 ] Josh Byster commented on ARTEMIS-4666: -- [~tabish] I tried that as well to no avail, same issue appears. > Federated queue consumers do not receive messages on classic clients > > > Key: ARTEMIS-4666 > URL: https://issues.apache.org/jira/browse/ARTEMIS-4666 > Project: ActiveMQ Artemis > Issue Type: Bug > Components: Federation >Affects Versions: 2.32.0 >Reporter: Josh Byster >Priority: Major > > Federated queues with A upstream and downstream to B do not seem to work as > expected when the client JMS implementation is ActiveMQ Classic v5.16.2 (used > in my example, but also verified the issue is present with v5.18.3 and > v6.0.1). With Artemis JMS as the client, it seems to work as expected. > When running a producer on A and a consumer on B with the classic > org.apache.activemq.ActiveMQConnectionFactory, the consumer on B does not > consume any messages that the producer sends. When B is restarted, it then > consumes the messages. > This works properly with ActiveMQ Artemis > org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory. I've > created a minimal reproducible example > [here|https://github.com/josh-byster/artemis-classic-consumer-bug/tree/master]. > Running server1 and server2 and then first starting up the Consumer then > running the Producer class, we can see that no messages are logged in console > by the consumer. When you restart the consumer, the messages are consumed. > This is an issue whether you attach a MessageListener or you call receive() > directly. If you switch the ActiveMQConnectionFactory implementation to > Artemis, it works as expected. > I don't think this necessarily warrants a fix if it's an issue specifically > with the classic client, since the solution is just to upgrade clients to > Artemis. However, if it is something that can be patched on the server, that > would be great. I do, however, think it would be good to note this down in > the docs that it's not supported with classic clients, since I spent a while > debugging it. However, most other features do work as expected with the > clients running the classic version, which is much appreciated as it makes > migration significantly easier. > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908359=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908359
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 15:50
Start Date: 05/Mar/24 15:50
Worklog Time Spent: 10m
Work Description: clebertsuconic commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513065063
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
Review Comment:
It still needs to create the file. which should happen away from the Netty
Thread.
in case of an exception it will initialize the exception on the session...
The current writes are still going through though but the exception would be
initialized. I could interrupt the addBytes if there was an exception, but the
possibility of having an exception there is actually so low that if any issues
are happening with the storage, other things will happen to put the server down
anyways through IOExceptions.
Issue Time Tracking
---
Worklog Id: (was: 908359)
Time Spent: 3h (was: 2h 50m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 3h
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908357=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908357
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 15:42
Start Date: 05/Mar/24 15:42
Worklog Time Spent: 10m
Work Description: tabish121 commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513036542
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
Review Comment:
This check seems to be mostly pointless now as the code doesn't use a
continuation to resume delivery after validation it just throws it into another
thread and then continues on as if it was successful
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPTunneledCoreMessageReader.java:
##
@@ -155,6 +149,10 @@ public Message readBytes(Delivery delivery) {
coreMessage.reloadPersistence(buffer,
sessionSPI.getCoreMessageObjectPools());
coreMessage.setMessageID(sessionSPI.getStorageManager().generateID());
- return coreMessage;
+ serverReceiver.onMessageComplete(delivery, coreMessage,
deliveryAnnotations);
+
+ close();
Review Comment:
The receiver should be closing this in onMessageComplete, a reader shouldn't
close itself.
Issue Time Tracking
---
Worklog Id: (was: 908357)
Time Spent: 2h 50m (was: 2h 40m)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 2h 50m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and file.sync should be moved away
> from the Netty Thread for AMQP Large Messages
> This task now is about moving the processing for AMQP Messages. we may in a
> near future also improve tunneled large messages. For now we will do for AMQP
> messages only.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4668) Move AMQP Large Message File Handling away from Netty thread
[
https://issues.apache.org/jira/browse/ARTEMIS-4668?focusedWorklogId=908353=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908353
]
ASF GitHub Bot logged work on ARTEMIS-4668:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 15:33
Start Date: 05/Mar/24 15:33
Worklog Time Spent: 10m
Work Description: tabish121 commented on code in PR #4840:
URL: https://github.com/apache/activemq-artemis/pull/4840#discussion_r1513034841
##
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPLargeMessageReader.java:
##
@@ -77,39 +73,73 @@ public AMQPLargeMessageReader open() {
}
@Override
- public Message readBytes(Delivery delivery) throws Exception {
+ public boolean readBytes(Delivery delivery) throws Exception {
if (closed) {
throw new IllegalStateException("AMQP Large Message Reader is closed
and read cannot proceed");
}
+ serverReceiver.connection.requireInHandler();
+
final Receiver receiver = ((Receiver) delivery.getLink());
final ReadableBuffer dataBuffer = receiver.recv();
+ final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
+
if (currentMessage == null) {
- final AMQPSessionCallback sessionSPI =
serverReceiver.getSessionContext().getSessionSPI();
final long id = sessionSPI.getStorageManager().generateID();
currentMessage = new AMQPLargeMessage(id,
delivery.getMessageFormat(), null,
sessionSPI.getCoreMessageObjectPools(),
sessionSPI.getStorageManager());
currentMessage.parseHeader(dataBuffer);
-
+ logger.trace("Initializing current message {} on {}", currentMessage,
this);
sessionSPI.getStorageManager().largeMessageCreated(id,
currentMessage);
+ sessionSPI.execute(() -> validateFile(currentMessage));
}
- currentMessage.addBytes(dataBuffer);
+ serverReceiver.getConnection().disableAutoRead();
- final AMQPLargeMessage result;
+ byte[] bytes = new byte[dataBuffer.remaining()];
+ dataBuffer.get(bytes);
- if (!delivery.isPartial()) {
-
currentMessage.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
- result = currentMessage;
- // We don't want a close to delete the file now, we've released the
resources.
- currentMessage = null;
- deliveryAnnotations = result.getDeliveryAnnotations();
- } else {
- result = null;
+ boolean partial = delivery.isPartial();
+
+ sessionSPI.execute(() -> addBytes(delivery, bytes, partial));
+
+ return partial;
+ }
+
+ private void validateFile(AMQPLargeMessage message) {
+ try {
+ message.validateFile();
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ close();
+ serverReceiver.connection.exception(e);
}
+ }
- return result;
+ private void addBytes(Delivery delivery, byte[] bytes, boolean isPartial) {
+ ReadableBuffer dataBuffer = ReadableBuffer.ByteBufferReader.wrap(bytes);
+ try {
+ logger.trace("Adding {} bytes on currentMessage={}, this={}",
dataBuffer.remaining(), currentMessage, this);
+ currentMessage.addBytes(dataBuffer);
+
+ if (!isPartial) {
+final AMQPLargeMessage message = currentMessage;
+
message.releaseResources(serverReceiver.getConnection().isLargeMessageSync(),
true);
+logger.trace("finishing {} on {}", currentMessage, this);
+// We don't want a close to delete the file now, we've released
the resources.
+currentMessage = null;
+close();
Review Comment:
This seems to be incorrect as closing here is wrong as a general rule as the
resource that opened this should be closing it but also this is occurring in a
different thread than the connection thread where the reader was opened which
could lead to error on inconsistent state. The receiver should be closing this
in the connection thread.
Issue Time Tracking
---
Worklog Id: (was: 908353)
Time Spent: 2h 40m (was: 2.5h)
> Move AMQP Large Message File Handling away from Netty thread
>
>
> Key: ARTEMIS-4668
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4668
> Project: ActiveMQ Artemis
> Issue Type: Bug
>Affects Versions: 2.32.0
>Reporter: Clebert Suconic
>Assignee: Clebert Suconic
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 2h 40m
> Remaining Estimate: 0h
>
> Operations like file.open, file.close, and
[jira] [Resolved] (AMQ-9419) UnsupportedOperationException("createContext() is not supported")
[
https://issues.apache.org/jira/browse/AMQ-9419?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jean-Baptiste Onofré resolved AMQ-9419.
---
Resolution: Fixed
> UnsupportedOperationException("createContext() is not supported")
> -
>
> Key: AMQ-9419
> URL: https://issues.apache.org/jira/browse/AMQ-9419
> Project: ActiveMQ Classic
> Issue Type: Bug
> Components: JMS client
>Affects Versions: 6.0.1
>Reporter: Terrien Jean-Yves
>Assignee: Jean-Baptiste Onofré
>Priority: Major
> Fix For: 6.1.0, 6.0.2
>
> Original Estimate: 1h
> Time Spent: 0.5h
> Remaining Estimate: 0.5h
>
> In ActiveMQ
> We have the "createContext" methods
> {code:java}
> createContext() => new ActiveMQContext(createActiveMQConnection())
> createContext(userName, password) => return new
> ActiveMQContext(createActiveMQConnection(userName, password))
> createContext(userName, password, sessionMode) => return new
> ActiveMQContext(createActiveMQConnection(userName, password), sessionMode)
> {code}
>
> But for
> {code:java}
> createContext(int sessionMode) => throw new
> UnsupportedOperationException("createContext() is not supported"){code}
> While
> {code:java}
> createContext(int sessionMode) => return new
> ActiveMQContext(createActiveMQConnection(getUserName(), getPassword()),
> sessionMode){code}
> works correctly
> Why raise this exception?
> Otherwise I suggest changing lines 327 to 332 of
> ActiveMQConnectionFactory.java
> by
> {code:java}
> public JMSContext createContext(String userName, String password, int
> sessionMode) {
> try {
> return new
> ActiveMQContext(createActiveMQConnection(getUserName(), getPassword()),
> sessionMode);
> } catch (JMSException e) {
> throw JMSExceptionSupport.convertToJMSRuntimeException(e);
> }
> }
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Resolved] (ARTEMIS-4655) Report logging metrics
[
https://issues.apache.org/jira/browse/ARTEMIS-4655?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Robbie Gemmell resolved ARTEMIS-4655.
-
Fix Version/s: 2.33.0
Resolution: Fixed
> Report logging metrics
> --
>
> Key: ARTEMIS-4655
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4655
> Project: ActiveMQ Artemis
> Issue Type: Improvement
>Reporter: Justin Bertram
>Assignee: Justin Bertram
>Priority: Major
> Fix For: 2.33.0
>
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
> External systems (e.g. Prometheus & Grafana) can consume metrics to visualize
> and monitor the health and performance of the broker. These systems often
> support configurable alerts to inform administrators of problems (e.g. long
> GC pauses, message accumulation, etc.). It may be useful to configure alerts
> for {{ERROR}} or {{WARN}} events in the log which may go unnoticed otherwise.
> The
> [Log4j2Metrics.java|https://github.com/micrometer-metrics/micrometer/blob/main/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/logging/Log4j2Metrics.java]
> provided by Micrometer will report metrics for each logging category so
> administrators can see when unexpected events occur.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4657) Support correlation ID compatibility between JMS clients
[
https://issues.apache.org/jira/browse/ARTEMIS-4657?focusedWorklogId=908286=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908286
]
ASF GitHub Bot logged work on ARTEMIS-4657:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 09:45
Start Date: 05/Mar/24 09:45
Worklog Time Spent: 10m
Work Description: gemmellr commented on code in PR #4833:
URL: https://github.com/apache/activemq-artemis/pull/4833#discussion_r1512508594
##
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java:
##
@@ -590,9 +590,9 @@ private static ActiveMQMessage
toAMQMessage(MessageReference reference,
}
amqMsg.setCommandId(commandId);
- final SimpleString corrId = getObjectProperty(coreMessage,
SimpleString.class, OpenWireConstants.JMS_CORRELATION_ID_PROPERTY);
- if (corrId != null) {
- amqMsg.setCorrelationId(corrId.toString());
+ final Object correlationID = coreMessage.getCorrelationID();
+ if (correlationID != null) {
+ amqMsg.setCorrelationId(correlationID.toString());
Review Comment:
I think the actual-expected-behaviour should be tested. This specific
behaviour wasnt noticed as there is a simply a 'disabled test as it wont send
matching bytes', which is not covering what actually was/is expected to
[not-]happen in this situation either before or after the changes here.
Issue Time Tracking
---
Worklog Id: (was: 908286)
Time Spent: 2h 10m (was: 2h)
> Support correlation ID compatibility between JMS clients
>
>
> Key: ARTEMIS-4657
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4657
> Project: ActiveMQ Artemis
> Issue Type: Improvement
>Reporter: Justin Bertram
>Assignee: Justin Bertram
>Priority: Major
> Time Spent: 2h 10m
> Remaining Estimate: 0h
>
> Currently there are some use-cases with both {{String}} and {{byte[]}} values
> of JMS correlation ID that don't work between Core, OpenWire, and AMQP. We
> should support as many as possible.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4657) Support correlation ID compatibility between JMS clients
[
https://issues.apache.org/jira/browse/ARTEMIS-4657?focusedWorklogId=908284=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908284
]
ASF GitHub Bot logged work on ARTEMIS-4657:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 09:44
Start Date: 05/Mar/24 09:44
Worklog Time Spent: 10m
Work Description: gemmellr commented on code in PR #4833:
URL: https://github.com/apache/activemq-artemis/pull/4833#discussion_r1512508594
##
artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java:
##
@@ -590,9 +590,9 @@ private static ActiveMQMessage
toAMQMessage(MessageReference reference,
}
amqMsg.setCommandId(commandId);
- final SimpleString corrId = getObjectProperty(coreMessage,
SimpleString.class, OpenWireConstants.JMS_CORRELATION_ID_PROPERTY);
- if (corrId != null) {
- amqMsg.setCorrelationId(corrId.toString());
+ final Object correlationID = coreMessage.getCorrelationID();
+ if (correlationID != null) {
+ amqMsg.setCorrelationId(correlationID.toString());
Review Comment:
I think the actual-expected-behaviour should be tested. This specific
behaviour wasnt noticed as there is a simply a 'disabled test as it wont send
matching bytes', which is not covering what actually was/is expected to happen
in this situation either before or after the changes here.
Issue Time Tracking
---
Worklog Id: (was: 908284)
Time Spent: 2h (was: 1h 50m)
> Support correlation ID compatibility between JMS clients
>
>
> Key: ARTEMIS-4657
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4657
> Project: ActiveMQ Artemis
> Issue Type: Improvement
>Reporter: Justin Bertram
>Assignee: Justin Bertram
>Priority: Major
> Time Spent: 2h
> Remaining Estimate: 0h
>
> Currently there are some use-cases with both {{String}} and {{byte[]}} values
> of JMS correlation ID that don't work between Core, OpenWire, and AMQP. We
> should support as many as possible.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (ARTEMIS-4655) Report logging metrics
[
https://issues.apache.org/jira/browse/ARTEMIS-4655?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17823517#comment-17823517
]
ASF subversion and git services commented on ARTEMIS-4655:
--
Commit 661a4e6fdc1d10b115c209a25e96e6b97e75e8e6 in activemq-artemis's branch
refs/heads/main from Justin Bertram
[ https://gitbox.apache.org/repos/asf?p=activemq-artemis.git;h=661a4e6fdc ]
ARTEMIS-4655 report logging metrics
It may be useful to configure alerts for ERROR or WARN events in the log
which may go unnoticed otherwise.
> Report logging metrics
> --
>
> Key: ARTEMIS-4655
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4655
> Project: ActiveMQ Artemis
> Issue Type: Improvement
>Reporter: Justin Bertram
>Assignee: Justin Bertram
>Priority: Major
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
> External systems (e.g. Prometheus & Grafana) can consume metrics to visualize
> and monitor the health and performance of the broker. These systems often
> support configurable alerts to inform administrators of problems (e.g. long
> GC pauses, message accumulation, etc.). It may be useful to configure alerts
> for {{ERROR}} or {{WARN}} events in the log which may go unnoticed otherwise.
> The
> [Log4j2Metrics.java|https://github.com/micrometer-metrics/micrometer/blob/main/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/logging/Log4j2Metrics.java]
> provided by Micrometer will report metrics for each logging category so
> administrators can see when unexpected events occur.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Work logged] (ARTEMIS-4655) Report logging metrics
[
https://issues.apache.org/jira/browse/ARTEMIS-4655?focusedWorklogId=908278=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-908278
]
ASF GitHub Bot logged work on ARTEMIS-4655:
---
Author: ASF GitHub Bot
Created on: 05/Mar/24 09:37
Start Date: 05/Mar/24 09:37
Worklog Time Spent: 10m
Work Description: gemmellr merged PR #4830:
URL: https://github.com/apache/activemq-artemis/pull/4830
Issue Time Tracking
---
Worklog Id: (was: 908278)
Time Spent: 1.5h (was: 1h 20m)
> Report logging metrics
> --
>
> Key: ARTEMIS-4655
> URL: https://issues.apache.org/jira/browse/ARTEMIS-4655
> Project: ActiveMQ Artemis
> Issue Type: Improvement
>Reporter: Justin Bertram
>Assignee: Justin Bertram
>Priority: Major
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
> External systems (e.g. Prometheus & Grafana) can consume metrics to visualize
> and monitor the health and performance of the broker. These systems often
> support configurable alerts to inform administrators of problems (e.g. long
> GC pauses, message accumulation, etc.). It may be useful to configure alerts
> for {{ERROR}} or {{WARN}} events in the log which may go unnoticed otherwise.
> The
> [Log4j2Metrics.java|https://github.com/micrometer-metrics/micrometer/blob/main/micrometer-core/src/main/java/io/micrometer/core/instrument/binder/logging/Log4j2Metrics.java]
> provided by Micrometer will report metrics for each logging category so
> administrators can see when unexpected events occur.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
