You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "MooseTheBrown (via GitHub)" <gi...@apache.org> on 2023/12/08 11:45:56 UTC
[PR] NIFI-6730 AMQP QoS support [nifi]
MooseTheBrown opened a new pull request, #8146:
URL: https://github.com/apache/nifi/pull/8146
<!-- 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. -->
# Summary
[NIFI-6730](https://issues.apache.org/jira/browse/NIFI-6730)
Implement basicQos (prefetch count) support in ConsumeAMQP processor.
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [x ] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [ x] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [ x] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
### Pull Request Formatting
- [ x] Pull Request based on current revision of the `main` branch
- [ x] Pull Request refers to a feature branch with one commit containing changes
# Verification
Please indicate the verification steps performed prior to pull request creation.
### Build
- [x ] Build completed using `mvn clean install -P contrib-check`
- [x ] JDK 21
### Licensing
- [ x] New dependencies are compatible with the [Apache License 2.0](https://apache.org/licenses/LICENSE-2.0) according to the [License Policy](https://www.apache.org/legal/resolved.html)
- [ x] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [ x] Documentation formatting appears as expected in rendered files
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "MooseTheBrown (via GitHub)" <gi...@apache.org>.
MooseTheBrown commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425103142
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
}
@Test
public void failOnNullQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, null, true, processorLog);
+ new AMQPConsumer(conn, null, true, 0, processorLog);
});
}
@Test
public void failOnEmptyQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, " ", true, processorLog);
+ new AMQPConsumer(conn, " ", true, 0, processorLog);
});
}
@Test
public void failOnNonExistingQueue() {
assertThrows(IOException.class, () -> {
Connection conn = new TestConnection(null, null);
- try (AMQPConsumer consumer = new AMQPConsumer(conn, "hello", true, processorLog)) {
+ try (AMQPConsumer consumer = new AMQPConsumer(conn, "hello", true, 0, processorLog)) {
Review Comment:
Fixed
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -117,7 +117,7 @@ public void validateSuccessfullConsumeWithEmptyQueueDefaultExchange() throws Exc
exchangeToRoutingKeymap.put("", "queue1");
Connection conn = new TestConnection(exchangeToRoutingKeymap, routingMap);
- try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, processorLog)) {
+ try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, 0, processorLog)) {
Review Comment:
Fixed
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -132,9 +132,20 @@ public void validateSuccessfullConsumeWithEmptyQueue() throws Exception {
Connection conn = new TestConnection(exchangeToRoutingKeymap, routingMap);
conn.createChannel().basicPublish("myExchange", "key1", null, "hello Joe".getBytes());
- try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, processorLog)) {
+ try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, 0, processorLog)) {
Review Comment:
Fixed
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "dan-s1 (via GitHub)" <gi...@apache.org>.
dan-s1 commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1424438916
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
}
@Test
public void failOnNullQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, null, true, processorLog);
+ new AMQPConsumer(conn, null, true, 0, processorLog);
});
}
@Test
public void failOnEmptyQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, " ", true, processorLog);
+ new AMQPConsumer(conn, " ", true, 0, processorLog);
Review Comment:
```suggestion
new AMQPConsumer(conn, " ", true, DEFAULT_PREFETCH_COUNT, processorLog);
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -69,7 +69,7 @@ public void testConsumerHandlesCancelling() throws IOException {
final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
final TestConnection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
- final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, processorLog);
+ final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, 0, processorLog);
Review Comment:
```suggestion
final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, DEFAULT_PREFETCH_COUNT, processorLog);
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -132,9 +132,20 @@ public void validateSuccessfullConsumeWithEmptyQueue() throws Exception {
Connection conn = new TestConnection(exchangeToRoutingKeymap, routingMap);
conn.createChannel().basicPublish("myExchange", "key1", null, "hello Joe".getBytes());
- try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, processorLog)) {
+ try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, 0, processorLog)) {
Review Comment:
```suggestion
try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, DEFAULT_PREFETCH_COUNT, processorLog)) {
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
}
@Test
public void failOnNullQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, null, true, processorLog);
+ new AMQPConsumer(conn, null, true, 0, processorLog);
Review Comment:
```suggestion
new AMQPConsumer(conn, null, true, DEFAULT_PREFETCH_COUNT, processorLog);
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -55,7 +55,7 @@ public void testResponseQueueDrained() throws TimeoutException, IOException {
final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
final TestConnection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
- final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, processorLog);
+ final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, 0, processorLog);
Review Comment:
Create a static final variable to clearly indicate the default prefetch count is being used e.g.
`private static final int DEFAULT_PREFETCH_COUNT = 0;`
similarly to what you do in `TestChannel` where you declare a variable for `prefetchCount`
```suggestion
final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, DEFAULT_PREFETCH_COUNT, processorLog);
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
Review Comment:
```suggestion
assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, DEFAULT_PREFETCH_COUNT, processorLog));
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -117,7 +117,7 @@ public void validateSuccessfullConsumeWithEmptyQueueDefaultExchange() throws Exc
exchangeToRoutingKeymap.put("", "queue1");
Connection conn = new TestConnection(exchangeToRoutingKeymap, routingMap);
- try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, processorLog)) {
+ try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, 0, processorLog)) {
Review Comment:
```suggestion
try (AMQPConsumer consumer = new AMQPConsumer(conn, "queue1", true, DEFAULT_PREFETCH_COUNT, processorLog)) {
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
}
@Test
public void failOnNullQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, null, true, processorLog);
+ new AMQPConsumer(conn, null, true, 0, processorLog);
});
}
@Test
public void failOnEmptyQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, " ", true, processorLog);
+ new AMQPConsumer(conn, " ", true, 0, processorLog);
});
}
@Test
public void failOnNonExistingQueue() {
assertThrows(IOException.class, () -> {
Connection conn = new TestConnection(null, null);
- try (AMQPConsumer consumer = new AMQPConsumer(conn, "hello", true, processorLog)) {
+ try (AMQPConsumer consumer = new AMQPConsumer(conn, "hello", true, 0, processorLog)) {
Review Comment:
```suggestion
try (AMQPConsumer consumer = new AMQPConsumer(conn, "hello", true, DEFAULT_PREFETCH_COUNT, processorLog)) {
```
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "asfgit (via GitHub)" <gi...@apache.org>.
asfgit closed pull request #8146: NIFI-6730 AMQP QoS support
URL: https://github.com/apache/nifi/pull/8146
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "dan-s1 (via GitHub)" <gi...@apache.org>.
dan-s1 commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425443615
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -43,6 +43,7 @@
public class AMQPConsumerTest {
private ComponentLog processorLog;
+ private static final int DEFAULT_PREFETCH_COUNT = 0;
Review Comment:
```suggestion
private static final int DEFAULT_PREFETCH_COUNT = 0;
private ComponentLog processorLog;
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -301,7 +316,8 @@ protected synchronized AMQPConsumer createAMQPWorker(final ProcessContext contex
try {
final String queueName = context.getProperty(QUEUE).getValue();
final boolean autoAcknowledge = context.getProperty(AUTO_ACKNOWLEDGE).asBoolean();
- final AMQPConsumer amqpConsumer = new AMQPConsumer(connection, queueName, autoAcknowledge, getLogger());
+ final AMQPConsumer amqpConsumer = new AMQPConsumer(connection, queueName, autoAcknowledge,
+ context.getProperty(PREFETCH_COUNT).asInteger(), getLogger());
Review Comment:
```suggestion
final int prefetchCount = context.getProperty(PREFETCH_COUNT).asInteger();
final AMQPConsumer amqpConsumer = new AMQPConsumer(connection, queueName, autoAcknowledge, prefetchCount, getLogger());
```
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "MooseTheBrown (via GitHub)" <gi...@apache.org>.
MooseTheBrown commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425102068
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +112,16 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .defaultValue("0")
+ .required(true)
+ .build();
Review Comment:
Added customValidate with logic to check prefetch count value to ConsumeAMQP since prefetch count is declared in its scope.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "dan-s1 (via GitHub)" <gi...@apache.org>.
dan-s1 commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1424500623
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +112,16 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .defaultValue("0")
+ .required(true)
+ .build();
Review Comment:
Per the [Rabbit javadocs](https://rabbitmq.github.io/rabbitmq-java-client/api/4.x.x/com/rabbitmq/client/Channel.html#basicQos) it seems the prefetch count must be an integer between 0 and 65535. You should add logic to `AbstractAMQPProcessor#customValidate` method in order to ensure the integer entered is between 0 and 65535 and not greater.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "MooseTheBrown (via GitHub)" <gi...@apache.org>.
MooseTheBrown commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1426414623
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +115,17 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it."
+ + "Allowed values: from 0 to 65535. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
Review Comment:
Corrected. Used standard long validator instead of customValidate in ConsumeAMQP
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "dan-s1 (via GitHub)" <gi...@apache.org>.
dan-s1 commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425453995
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +112,16 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .defaultValue("0")
+ .required(true)
+ .build();
Review Comment:
You are correct. I am sorry about that, I am not sure what i was thinking when I entered `AbstractAMQPProcessor`
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "MooseTheBrown (via GitHub)" <gi...@apache.org>.
MooseTheBrown commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1426806679
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +115,17 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it."
+ + "Allowed values: from 0 to 65535. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
Review Comment:
Yes, thanks for pointing this out. Removed unused imports.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "dan-s1 (via GitHub)" <gi...@apache.org>.
dan-s1 commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1424437510
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -55,7 +55,7 @@ public void testResponseQueueDrained() throws TimeoutException, IOException {
final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
final TestConnection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
- final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, processorLog);
+ final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, 0, processorLog);
Review Comment:
Create a static final variable to clearly indicate the default prefetch count is being used e.g.
`private static final int DEFAULT_PREFETCH_COUNT = 0;`
similarly to what you did in `TestChannel` where you declared a variable for `prefetchCount`
```suggestion
final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, DEFAULT_PREFETCH_COUNT, processorLog);
```
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "MooseTheBrown (via GitHub)" <gi...@apache.org>.
MooseTheBrown commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425102583
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -55,7 +55,7 @@ public void testResponseQueueDrained() throws TimeoutException, IOException {
final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
final TestConnection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
- final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, processorLog);
+ final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, 0, processorLog);
Review Comment:
Fixed.
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -69,7 +69,7 @@ public void testConsumerHandlesCancelling() throws IOException {
final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
final TestConnection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
- final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, processorLog);
+ final AMQPConsumer consumer = new AMQPConsumer(connection, "queue1", true, 0, processorLog);
Review Comment:
Fixed
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1426750707
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +115,17 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it."
+ + "Allowed values: from 0 to 65535. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
Review Comment:
Thanks @MooseTheBrown! Could you please also remove the now unused imports? The build fails with Checkstyle violations. You can check it with running `mvn clean verify -P contrib-check` in `nifi-amqp-bundle` directory.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "MooseTheBrown (via GitHub)" <gi...@apache.org>.
MooseTheBrown commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425102857
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
Review Comment:
Fixed
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
}
@Test
public void failOnNullQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, null, true, processorLog);
+ new AMQPConsumer(conn, null, true, 0, processorLog);
Review Comment:
Fixed
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/AMQPConsumerTest.java:
##########
@@ -80,30 +80,30 @@ public void testConsumerHandlesCancelling() throws IOException {
@Test
public void failOnNullConnection() {
- assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, processorLog));
+ assertThrows(IllegalArgumentException.class, () -> new AMQPConsumer(null, null, true, 0, processorLog));
}
@Test
public void failOnNullQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, null, true, processorLog);
+ new AMQPConsumer(conn, null, true, 0, processorLog);
});
}
@Test
public void failOnEmptyQueueName() {
assertThrows(IllegalArgumentException.class, () -> {
Connection conn = new TestConnection(null, null);
- new AMQPConsumer(conn, " ", true, processorLog);
+ new AMQPConsumer(conn, " ", true, 0, processorLog);
Review Comment:
Fixed
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "dan-s1 (via GitHub)" <gi...@apache.org>.
dan-s1 commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1424500623
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +112,16 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .defaultValue("0")
+ .required(true)
+ .build();
Review Comment:
Per the [Rabbit javadocs](https://rabbitmq.github.io/rabbitmq-java-client/api/4.x.x/com/rabbitmq/client/Channel.html#basicQos) it seems the prefetch count must be an integer between 0 and 65535. You should add logic to `AbstractAMQPProcessor#customValidate` in order to ensure the integer entered is between 0 and 65535 and not greater.
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] NIFI-6730 AMQP QoS support [nifi]
Posted by "turcsanyip (via GitHub)" <gi...@apache.org>.
turcsanyip commented on code in PR #8146:
URL: https://github.com/apache/nifi/pull/8146#discussion_r1425813999
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -112,6 +115,17 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+ static final PropertyDescriptor PREFETCH_COUNT = new PropertyDescriptor.Builder()
+ .name("prefetch.count")
+ .displayName("Prefetch Count")
+ .description("The maximum number of unacknowledged messages for the consumer. If consumer has this number of unacknowledged messages, AMQP broker will "
+ + "no longer send new messages until consumer acknowledges some of the messages already delivered to it."
+ + "Allowed values: from 0 to 65535. 0 means no limit")
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
Review Comment:
@MooseTheBrown Thanks for adding `Prefetch Count` property!
Please note, there is a standard validator for checking if a number is within a range: https://github.com/apache/nifi/blob/5b664147eeccd9d2a8ded9040a3d64c96e70d639/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/processor/util/StandardValidators.java#L758
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org