You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "umarhussain15 (via GitHub)" <gi...@apache.org> on 2023/08/27 18:33:46 UTC
[GitHub] [nifi] umarhussain15 opened a new pull request, #7652: NIFI-11995 Support multiple output formats for message headers in ConsumeAMQP
umarhussain15 opened a new pull request, #7652:
URL: https://github.com/apache/nifi/pull/7652
<!-- 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-11995](https://issues.apache.org/jira/browse/NIFI-11995) extends ConsumeAMQP processor and allows the user to configure how to output headers from a received message. It provides three options:
* Output headers to `amqp$headers` attribute as string joined by the provided separator (Default) -> backward compatible
* Output headers to `amqp$headers` attribute as a valid JSON string. Headers with null value will be present as well in json with null value
* Output each header key as `<Header Key Prefix>.<header key>` attribute. It will not put headers which are null in message
* Property `<Header Key Prefix>` specifies the prefix value, defaults to `consume.amqp`
Test cases updated for new changes to the 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 17
### 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-11995 Support multiple output formats for message headers in ConsumeAMQP [nifi]
Posted by "umarhussain15 (via GitHub)" <gi...@apache.org>.
umarhussain15 commented on PR #7652:
URL: https://github.com/apache/nifi/pull/7652#issuecomment-1771413327
Hi @nandorsoma /@exceptionfactory, the ci-workflow failed for the MacOS build due to java heap space issue. Do I need to make some changes, or will you re-run the failed stage?
--
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
[GitHub] [nifi] umarhussain15 commented on pull request #7652: NIFI-11995 Support multiple output formats for message headers in ConsumeAMQP
Posted by "umarhussain15 (via GitHub)" <gi...@apache.org>.
umarhussain15 commented on PR #7652:
URL: https://github.com/apache/nifi/pull/7652#issuecomment-1698902191
Thanks, @nandorsoma, I have updated the pull request.
Can you point me to a resource to configure Java code formatter in IntelliJ IDEA for NiFi project?
My current formatter (google-java-format) is using 4 spaces for indentation, I think which changes the formatting in code.
--
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-11995 Support multiple output formats for message headers in ConsumeAMQP [nifi]
Posted by "umarhussain15 (via GitHub)" <gi...@apache.org>.
umarhussain15 commented on PR #7652:
URL: https://github.com/apache/nifi/pull/7652#issuecomment-1793856291
Thanks, @exceptionfactory. The suggestions look good, and I have applied them to the PR.
--
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-11995 Support multiple output formats for message headers in ConsumeAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory closed pull request #7652: NIFI-11995 Support multiple output formats for message headers in ConsumeAMQP
URL: https://github.com/apache/nifi/pull/7652
--
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-11995 Support multiple output formats for message headers in ConsumeAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7652:
URL: https://github.com/apache/nifi/pull/7652#discussion_r1348157253
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -214,14 +264,23 @@ private void addAttribute(final Map<String, String> attributes, final String att
attributes.put(attributeName, value.toString());
}
- private String buildHeaders(Map<String, Object> headers, boolean removeCurlyBraces, String valueSeparatorForHeaders) {
+ private String buildHeaders(Map<String, Object> headers, AllowableValue headerFormat, boolean removeCurlyBraces, String valueSeparatorForHeaders) {
if (headers == null) {
return null;
}
- String headerString = convertMapToString(headers,valueSeparatorForHeaders);
+ String headerString = null;
+ if (headerFormat.equals(HEADERS_FORMAT_COMMA_SEPARATED_STRING)) {
+ headerString = convertMapToString(headers, valueSeparatorForHeaders);
- if (!removeCurlyBraces) {
- headerString = "{" + headerString + "}";
+ if (!removeCurlyBraces) {
+ headerString = "{" + headerString + "}";
+ }
+ } else if (headerFormat.equals(HEADERS_FORMAT_JSON_STRING)) {
+ try {
+ headerString = convertMapToJSONString(headers);
+ } catch (JsonProcessingException e) {
+ getLogger().debug("error creating json string", e);
Review Comment:
It seems like this should be logged as a warning instead of debug.
```suggestion
getLogger().warn("Header formatting as JSON failed", e);
```
--
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-11995 Support multiple output formats for message headers in ConsumeAMQP [nifi]
Posted by "exceptionfactory (via GitHub)" <gi...@apache.org>.
exceptionfactory commented on code in PR #7652:
URL: https://github.com/apache/nifi/pull/7652#discussion_r1379386210
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -68,15 +72,24 @@
@WritesAttribute(attribute = "amqp$exchange", description = "The exchange from which AMQP Message was received")
})
public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
+
private static final String ATTRIBUTES_PREFIX = "amqp$";
+ public static final String DEFAULT_HEADERS_KEY_PREFIX = "consume.amqp";
+
+ public static final AllowableValue HEADERS_FORMAT_COMMA_SEPARATED_STRING = new AllowableValue("Comma Separated String", "Comma Separated String",
Review Comment:
```suggestion
public static final AllowableValue HEADERS_FORMAT_COMMA_SEPARATED_STRING = new AllowableValue("Comma-Separated String", "Comma-Separated String",
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -68,15 +72,24 @@
@WritesAttribute(attribute = "amqp$exchange", description = "The exchange from which AMQP Message was received")
})
public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
+
private static final String ATTRIBUTES_PREFIX = "amqp$";
+ public static final String DEFAULT_HEADERS_KEY_PREFIX = "consume.amqp";
+
+ public static final AllowableValue HEADERS_FORMAT_COMMA_SEPARATED_STRING = new AllowableValue("Comma Separated String", "Comma Separated String",
+ "Put all headers as a string with the specified separator in the attribute 'amqp$headers'.");
+ public static final AllowableValue HEADERS_FORMAT_JSON_STRING = new AllowableValue("JSON String", "JSON String",
+ "Format all headers as JSON string and output in the attribute 'amqp$headers'. It will include keys with null value as well.");
+ public static final AllowableValue HEADERS_FORMAT_ATTRIBUTES = new AllowableValue("Flow file attributes", "Flow file attributes",
Review Comment:
```suggestion
public static final AllowableValue HEADERS_FORMAT_ATTRIBUTES = new AllowableValue("FlowFile Attributes", "FlowFile Attributes",
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/ConsumeAMQPTest.java:
##########
@@ -156,13 +158,83 @@ public void validateSuccessfulConsumeAndTransferToSuccess() throws Exception {
TestRunner runner = initTestRunner(proc);
runner.run();
- final MockFlowFile successFF = runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).get(0);
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
assertNotNull(successFF);
successFF.assertAttributeEquals("amqp$routingKey", "key1");
successFF.assertAttributeEquals("amqp$exchange", "myExchange");
}
}
+ @Test
+ public void validateHeaderWithJsonStringForHeaderFormatParameterConsumeAndTransferToSuccess() throws Exception {
+ final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2"));
+ final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
+ final Map<String, Object> headersMap = new HashMap<>();
+ headersMap.put("foo1", "bar,bar");
+ headersMap.put("foo2", "bar,bar");
+ headersMap.put("foo3", "null");
+ headersMap.put("foo4", null);
+ ObjectMapper objectMapper = new ObjectMapper();
+ JsonNode EXPECTED_JSON = objectMapper.valueToTree(headersMap);
+
+ AMQP.BasicProperties.Builder builderBasicProperties = new AMQP.BasicProperties.Builder();
+ builderBasicProperties.headers(headersMap);
+
+ final Connection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
+
+ try (AMQPPublisher sender = new AMQPPublisher(connection, mock(ComponentLog.class))) {
+ sender.publish("hello".getBytes(), builderBasicProperties.build(), "key1", "myExchange");
+
+ ConsumeAMQP proc = new LocalConsumeAMQP(connection);
+ TestRunner runner = initTestRunner(proc);
+ runner.setProperty(ConsumeAMQP.HEADER_FORMAT, ConsumeAMQP.HEADERS_FORMAT_JSON_STRING);
+ runner.run();
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
+ assertNotNull(successFF);
+ successFF.assertAttributeEquals("amqp$routingKey", "key1");
+ successFF.assertAttributeEquals("amqp$exchange", "myExchange");
+ String headers = successFF.getAttribute("amqp$headers");
+ JsonNode jsonNode = objectMapper.readTree(headers);
+ assertEquals(EXPECTED_JSON, jsonNode);
Review Comment:
```suggestion
assertEquals(expectedJson, jsonNode);
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/ConsumeAMQPTest.java:
##########
@@ -156,13 +158,83 @@ public void validateSuccessfulConsumeAndTransferToSuccess() throws Exception {
TestRunner runner = initTestRunner(proc);
runner.run();
- final MockFlowFile successFF = runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).get(0);
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
assertNotNull(successFF);
successFF.assertAttributeEquals("amqp$routingKey", "key1");
successFF.assertAttributeEquals("amqp$exchange", "myExchange");
}
}
+ @Test
+ public void validateHeaderWithJsonStringForHeaderFormatParameterConsumeAndTransferToSuccess() throws Exception {
+ final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2"));
+ final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
+ final Map<String, Object> headersMap = new HashMap<>();
+ headersMap.put("foo1", "bar,bar");
+ headersMap.put("foo2", "bar,bar");
+ headersMap.put("foo3", "null");
+ headersMap.put("foo4", null);
+ ObjectMapper objectMapper = new ObjectMapper();
+ JsonNode EXPECTED_JSON = objectMapper.valueToTree(headersMap);
+
+ AMQP.BasicProperties.Builder builderBasicProperties = new AMQP.BasicProperties.Builder();
+ builderBasicProperties.headers(headersMap);
+
+ final Connection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
+
+ try (AMQPPublisher sender = new AMQPPublisher(connection, mock(ComponentLog.class))) {
+ sender.publish("hello".getBytes(), builderBasicProperties.build(), "key1", "myExchange");
+
+ ConsumeAMQP proc = new LocalConsumeAMQP(connection);
+ TestRunner runner = initTestRunner(proc);
+ runner.setProperty(ConsumeAMQP.HEADER_FORMAT, ConsumeAMQP.HEADERS_FORMAT_JSON_STRING);
+ runner.run();
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
+ assertNotNull(successFF);
+ successFF.assertAttributeEquals("amqp$routingKey", "key1");
+ successFF.assertAttributeEquals("amqp$exchange", "myExchange");
+ String headers = successFF.getAttribute("amqp$headers");
+ JsonNode jsonNode = objectMapper.readTree(headers);
+ assertEquals(EXPECTED_JSON, jsonNode);
+ }
+ }
+
+ @Test
+ public void validateHeaderWithFlowFileAttributeForHeaderFormatParameterConsumeAndTransferToSuccess() throws Exception {
+ final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2"));
+ final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
+ final Map<String,Object> expectedHeadersMap = new HashMap<>();
+ expectedHeadersMap.put("foo1", "bar,bar");
+ expectedHeadersMap.put("foo2", "bar,bar");
+ expectedHeadersMap.put("foo3", "null");
+ final Map<String, Object> headersMap = new HashMap<>(expectedHeadersMap);
+ headersMap.put("foo4", null);
+
+ final String HEADER_PREFIX = "test.header";
Review Comment:
```suggestion
final String headerPrefix = "test.header";
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -170,8 +208,8 @@ protected void processResource(final Connection connection, final AMQPConsumer c
final BasicProperties amqpProperties = response.getProps();
final Envelope envelope = response.getEnvelope();
- final Map<String, String> attributes = buildAttributes(amqpProperties, envelope, context.getProperty(REMOVE_CURLY_BRACES).asBoolean(),
- context.getProperty(HEADER_SEPARATOR).toString());
+ final Map<String, String> attributes = buildAttributes(amqpProperties, envelope, context.getProperty(HEADER_FORMAT).toString(), context.getProperty(HEADER_KEY_PREFIX).toString(),
Review Comment:
It would be helpful to declare the values for `headerFormat` and `headerKeyPrefix`. In addition, `PropertyValue.getValue()` should be used instead of `PropertyValue.toString()`:
```suggestion
final String headerFormat = context.getProperty(HEADER_FORMAT).getValue();
final String headerKeyPrefix = context.getProperty(HEADER_KEY_PREFIX).getValue();
final Map<String, String> attributes = buildAttributes(amqpProperties, envelope, headerFormat, headerKeyPrefix,
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -99,22 +112,43 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+
+ public static final PropertyDescriptor HEADER_FORMAT = new PropertyDescriptor.Builder()
+ .name("header.format")
+ .displayName("Header Output Format")
+ .description("Defines how to output headers from the received message")
+ .allowableValues(HEADERS_FORMAT_COMMA_SEPARATED_STRING, HEADERS_FORMAT_JSON_STRING, HEADERS_FORMAT_ATTRIBUTES)
+ .defaultValue(HEADERS_FORMAT_COMMA_SEPARATED_STRING.getValue())
+ .required(true)
+ .build();
+ public static final PropertyDescriptor HEADER_KEY_PREFIX = new PropertyDescriptor.Builder()
+ .name("header.key.prefix")
+ .displayName("Header Key Prefix")
+ .description("Text to be prefixed to header keys as the are added to the flowfile attributes. Processor will append '.' to the value")
Review Comment:
```suggestion
.description("Text to be prefixed to header keys as the are added to the FlowFile attributes. Processor will append '.' to the value of this property")
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -99,22 +112,43 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+
+ public static final PropertyDescriptor HEADER_FORMAT = new PropertyDescriptor.Builder()
+ .name("header.format")
+ .displayName("Header Output Format")
+ .description("Defines how to output headers from the received message")
+ .allowableValues(HEADERS_FORMAT_COMMA_SEPARATED_STRING, HEADERS_FORMAT_JSON_STRING, HEADERS_FORMAT_ATTRIBUTES)
+ .defaultValue(HEADERS_FORMAT_COMMA_SEPARATED_STRING.getValue())
+ .required(true)
+ .build();
+ public static final PropertyDescriptor HEADER_KEY_PREFIX = new PropertyDescriptor.Builder()
+ .name("header.key.prefix")
+ .displayName("Header Key Prefix")
+ .description("Text to be prefixed to header keys as the are added to the flowfile attributes. Processor will append '.' to the value")
+ .defaultValue(DEFAULT_HEADERS_KEY_PREFIX)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .dependsOn(HEADER_FORMAT, HEADERS_FORMAT_ATTRIBUTES)
+ .required(true)
+ .build();
+
public static final PropertyDescriptor HEADER_SEPARATOR = new PropertyDescriptor.Builder()
- .name("header.separator")
- .displayName("Header Separator")
- .description("The character that is used to separate key-value for header in String. The value must only one character."
- + "Otherwise you will get an error message")
- .addValidator(StandardValidators.SINGLE_CHAR_VALIDATOR)
- .defaultValue(",")
- .required(false)
- .build();
+ .name("header.separator")
+ .displayName("Header Separator")
+ .description("The character that is used to separate key-value for header in String. The value must only one character."
+ + "Otherwise you will get an error message")
Review Comment:
The use of `you` should be avoided, and invalid values generally produce errors, so this can be removed.
```suggestion
)
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/ConsumeAMQPTest.java:
##########
@@ -156,13 +158,83 @@ public void validateSuccessfulConsumeAndTransferToSuccess() throws Exception {
TestRunner runner = initTestRunner(proc);
runner.run();
- final MockFlowFile successFF = runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).get(0);
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
assertNotNull(successFF);
successFF.assertAttributeEquals("amqp$routingKey", "key1");
successFF.assertAttributeEquals("amqp$exchange", "myExchange");
}
}
+ @Test
+ public void validateHeaderWithJsonStringForHeaderFormatParameterConsumeAndTransferToSuccess() throws Exception {
+ final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2"));
+ final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
+ final Map<String, Object> headersMap = new HashMap<>();
+ headersMap.put("foo1", "bar,bar");
+ headersMap.put("foo2", "bar,bar");
+ headersMap.put("foo3", "null");
+ headersMap.put("foo4", null);
+ ObjectMapper objectMapper = new ObjectMapper();
+ JsonNode EXPECTED_JSON = objectMapper.valueToTree(headersMap);
Review Comment:
Local variables should be camelCased.
```suggestion
JsonNode expectedJson = objectMapper.valueToTree(headersMap);
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/main/java/org/apache/nifi/amqp/processors/ConsumeAMQP.java:
##########
@@ -99,22 +112,43 @@ public class ConsumeAMQP extends AbstractAMQPProcessor<AMQPConsumer> {
.defaultValue("10")
.required(true)
.build();
+
+ public static final PropertyDescriptor HEADER_FORMAT = new PropertyDescriptor.Builder()
+ .name("header.format")
+ .displayName("Header Output Format")
+ .description("Defines how to output headers from the received message")
+ .allowableValues(HEADERS_FORMAT_COMMA_SEPARATED_STRING, HEADERS_FORMAT_JSON_STRING, HEADERS_FORMAT_ATTRIBUTES)
+ .defaultValue(HEADERS_FORMAT_COMMA_SEPARATED_STRING.getValue())
+ .required(true)
+ .build();
+ public static final PropertyDescriptor HEADER_KEY_PREFIX = new PropertyDescriptor.Builder()
+ .name("header.key.prefix")
+ .displayName("Header Key Prefix")
+ .description("Text to be prefixed to header keys as the are added to the flowfile attributes. Processor will append '.' to the value")
+ .defaultValue(DEFAULT_HEADERS_KEY_PREFIX)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .dependsOn(HEADER_FORMAT, HEADERS_FORMAT_ATTRIBUTES)
+ .required(true)
+ .build();
+
public static final PropertyDescriptor HEADER_SEPARATOR = new PropertyDescriptor.Builder()
- .name("header.separator")
- .displayName("Header Separator")
- .description("The character that is used to separate key-value for header in String. The value must only one character."
- + "Otherwise you will get an error message")
- .addValidator(StandardValidators.SINGLE_CHAR_VALIDATOR)
- .defaultValue(",")
- .required(false)
- .build();
+ .name("header.separator")
+ .displayName("Header Separator")
+ .description("The character that is used to separate key-value for header in String. The value must only one character."
Review Comment:
```suggestion
.description("The character that is used to separate key-value for header in String. The value must be only one character."
```
##########
nifi-nar-bundles/nifi-amqp-bundle/nifi-amqp-processors/src/test/java/org/apache/nifi/amqp/processors/ConsumeAMQPTest.java:
##########
@@ -156,13 +158,83 @@ public void validateSuccessfulConsumeAndTransferToSuccess() throws Exception {
TestRunner runner = initTestRunner(proc);
runner.run();
- final MockFlowFile successFF = runner.getFlowFilesForRelationship(PublishAMQP.REL_SUCCESS).get(0);
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
assertNotNull(successFF);
successFF.assertAttributeEquals("amqp$routingKey", "key1");
successFF.assertAttributeEquals("amqp$exchange", "myExchange");
}
}
+ @Test
+ public void validateHeaderWithJsonStringForHeaderFormatParameterConsumeAndTransferToSuccess() throws Exception {
+ final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2"));
+ final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
+ final Map<String, Object> headersMap = new HashMap<>();
+ headersMap.put("foo1", "bar,bar");
+ headersMap.put("foo2", "bar,bar");
+ headersMap.put("foo3", "null");
+ headersMap.put("foo4", null);
+ ObjectMapper objectMapper = new ObjectMapper();
+ JsonNode EXPECTED_JSON = objectMapper.valueToTree(headersMap);
+
+ AMQP.BasicProperties.Builder builderBasicProperties = new AMQP.BasicProperties.Builder();
+ builderBasicProperties.headers(headersMap);
+
+ final Connection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
+
+ try (AMQPPublisher sender = new AMQPPublisher(connection, mock(ComponentLog.class))) {
+ sender.publish("hello".getBytes(), builderBasicProperties.build(), "key1", "myExchange");
+
+ ConsumeAMQP proc = new LocalConsumeAMQP(connection);
+ TestRunner runner = initTestRunner(proc);
+ runner.setProperty(ConsumeAMQP.HEADER_FORMAT, ConsumeAMQP.HEADERS_FORMAT_JSON_STRING);
+ runner.run();
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
+ assertNotNull(successFF);
+ successFF.assertAttributeEquals("amqp$routingKey", "key1");
+ successFF.assertAttributeEquals("amqp$exchange", "myExchange");
+ String headers = successFF.getAttribute("amqp$headers");
+ JsonNode jsonNode = objectMapper.readTree(headers);
+ assertEquals(EXPECTED_JSON, jsonNode);
+ }
+ }
+
+ @Test
+ public void validateHeaderWithFlowFileAttributeForHeaderFormatParameterConsumeAndTransferToSuccess() throws Exception {
+ final Map<String, List<String>> routingMap = Collections.singletonMap("key1", Arrays.asList("queue1", "queue2"));
+ final Map<String, String> exchangeToRoutingKeymap = Collections.singletonMap("myExchange", "key1");
+ final Map<String,Object> expectedHeadersMap = new HashMap<>();
+ expectedHeadersMap.put("foo1", "bar,bar");
+ expectedHeadersMap.put("foo2", "bar,bar");
+ expectedHeadersMap.put("foo3", "null");
+ final Map<String, Object> headersMap = new HashMap<>(expectedHeadersMap);
+ headersMap.put("foo4", null);
+
+ final String HEADER_PREFIX = "test.header";
+
+ AMQP.BasicProperties.Builder builderBasicProperties = new AMQP.BasicProperties.Builder();
+ builderBasicProperties.headers(headersMap);
+
+ final Connection connection = new TestConnection(exchangeToRoutingKeymap, routingMap);
+
+ try (AMQPPublisher sender = new AMQPPublisher(connection, mock(ComponentLog.class))) {
+ sender.publish("hello".getBytes(), builderBasicProperties.build(), "key1", "myExchange");
+
+ ConsumeAMQP proc = new LocalConsumeAMQP(connection);
+ TestRunner runner = initTestRunner(proc);
+ runner.setProperty(ConsumeAMQP.HEADER_FORMAT, ConsumeAMQP.HEADERS_FORMAT_ATTRIBUTES);
+ runner.setProperty(ConsumeAMQP.HEADER_KEY_PREFIX,HEADER_PREFIX);
+ runner.run();
+ final MockFlowFile successFF = runner.getFlowFilesForRelationship(ConsumeAMQP.REL_SUCCESS).get(0);
+ assertNotNull(successFF);
+ successFF.assertAttributeEquals("amqp$routingKey", "key1");
+ successFF.assertAttributeEquals("amqp$exchange", "myExchange");
+ successFF.assertAttributeNotExists("amqp$headers");
+ expectedHeadersMap.forEach((key, value) ->{
+ successFF.assertAttributeEquals(HEADER_PREFIX+"."+key,value.toString());
Review Comment:
The spacing should be adjusted.
```suggestion
successFF.assertAttributeEquals(HEADER_PREFIX + "." + key, value.toString());
```
--
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-11995 Support multiple output formats for message headers in ConsumeAMQP [nifi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #7652:
URL: https://github.com/apache/nifi/pull/7652#issuecomment-1802843410
## [Codecov](https://app.codecov.io/gh/apache/nifi/pull/7652?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> :exclamation: No coverage uploaded for pull request base (`main@eb6a5a2`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#section-missing-base-commit).
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## main #7652 +/- ##
=======================================
Coverage ? 50.60%
Complexity ? 35103
=======================================
Files ? 5251
Lines ? 275605
Branches ? 30435
=======================================
Hits ? 139476
Misses ? 126049
Partials ? 10080
```
:mega: Codecov offers a browser extension for seamless coverage viewing on GitHub. Try it in [Chrome](https://chrome.google.com/webstore/detail/codecov/gedikamndpbemklijjkncpnolildpbgo) or [Firefox](https://addons.mozilla.org/en-US/firefox/addon/codecov/) today!
--
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
[GitHub] [nifi] nandorsoma commented on pull request #7652: NIFI-11995 Support multiple output formats for message headers in ConsumeAMQP
Posted by "nandorsoma (via GitHub)" <gi...@apache.org>.
nandorsoma commented on PR #7652:
URL: https://github.com/apache/nifi/pull/7652#issuecomment-1705731932
Thanks @umarhussain15. Unfortunately, we have work to do on this topic. We have a pretty minimal `checkstyle.xml` in our root folder. In its current form, it is not enough to import using the checkstyle plugin in IDEA that you would typically do. So, the goal is to create a custom format that honors the checkstyle rules and doesn't force reformating existing code. You should be good if you use IntelliJ IDEA's `Default` with a little tweak setting the max line length to 200.
--
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