You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2022/10/14 13:57:44 UTC
[GitHub] [nifi] nandorsoma opened a new pull request, #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
nandorsoma opened a new pull request, #6530:
URL: https://github.com/apache/nifi/pull/6530
<!-- 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-10644](https://issues.apache.org/jira/browse/NIFI-10644)
Similar to ConsumeMQTT, this pr implements Message Demarcator for processing new-line (or other character sequence) delimited data.
# 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 8
- [ ] JDK 11
- [ ] JDK 17
### Licensing
- [ ] 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)
- [ ] New dependencies are documented in applicable `LICENSE` and `NOTICE` files
### Documentation
- [ ] 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
[GitHub] [nifi] nandorsoma commented on a diff in pull request #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
Posted by GitBox <gi...@apache.org>.
nandorsoma commented on code in PR #6530:
URL: https://github.com/apache/nifi/pull/6530#discussion_r996491653
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/PublishMQTT.java:
##########
@@ -109,6 +113,15 @@ public class PublishMQTT extends AbstractMQTTProcessor implements MqttCallback {
.description("The Record Writer to use for serializing Records before publishing them as an MQTT Message.")
.build();
+ public static final PropertyDescriptor MESSAGE_DEMARCATOR = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(BASE_MESSAGE_DEMARCATOR)
+ .description("With this property, you have an option to publish multiple messages from a single FlowFile. "
+ + "This property allows you to provide a string (interpreted as UTF-8) to use for splitting "
+ + "the FlowFile content. This is an optional property ; if not provided, and if not defining a "
+ + "Record Reader/Writer, each message received will result in a single FlowFile. To enter special "
Review Comment:
Oh, I tried to match the description to the processor, but seems like I missed this sentence. Thx. I also reverted the "splitting" change. It looked okay when I changed it, but now I think it makes more sense to use the same wording as ConsumeMQTT.
--
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] asfgit closed pull request #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
URL: https://github.com/apache/nifi/pull/6530
--
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] turcsanyip commented on a diff in pull request #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
Posted by GitBox <gi...@apache.org>.
turcsanyip commented on code in PR #6530:
URL: https://github.com/apache/nifi/pull/6530#discussion_r996398492
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/common/AbstractMQTTProcessor.java:
##########
@@ -291,6 +299,12 @@ public Collection<ValidationResult> customValidate(final ValidationContext valid
.explanation("both properties must be set when used.").build());
}
+ final boolean demarcatorIsSet = validationContext.getProperty(BASE_MESSAGE_DEMARCATOR).isSet();
+ if (readerIsSet && demarcatorIsSet) {
+ results.add(new ValidationResult.Builder().subject("Reader and Writer").valid(false)
+ .explanation("message Demarcator and Record Reader/Writer cannot be used at the same time.").build());
Review Comment:
Old typo:
```suggestion
.explanation("Message Demarcator and Record Reader/Writer cannot be used at the same time.").build());
```
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/test/java/org/apache/nifi/processors/mqtt/TestPublishMQTT.java:
##########
@@ -70,6 +73,18 @@ public void cleanup() {
mqttTestClient = null;
}
+ @Test
+ public void testRecordAndDemarcatorConfigurationTogetherIsInvalid() throws InitializationException {
+ mqttTestClient = new MqttTestClient(MqttTestClient.ConnectType.Subscriber);
Review Comment:
Copy-paste error:
```suggestion
mqttTestClient = new MqttTestClient(MqttTestClient.ConnectType.Publisher);
```
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/ConsumeMQTT.java:
##########
@@ -230,8 +225,8 @@ public class ConsumeMQTT extends AbstractMQTTProcessor implements MqttCallback {
PROP_QOS,
RECORD_READER,
RECORD_WRITER,
- ADD_ATTRIBUTES_AS_FIELDS,
MESSAGE_DEMARCATOR,
+ ADD_ATTRIBUTES_AS_FIELDS,
Review Comment:
`Add attributes as fields` property belongs to the Record Reader/Writer and is visible only when a Record Reader is configured. So I would not change the order.
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/PublishMQTT.java:
##########
@@ -336,4 +315,110 @@ public void deliveryComplete(String token) {
logger.trace("Received 'delivery complete' message from broker. Token: [{}]", token);
}
+ interface ProcessStrategy {
+ void process(ProcessContext context, FlowFile flowfile, InputStream in, String topic, AtomicInteger processedRecords, Long previousProcessFailedAt) throws IOException;
+ String getFailureTemplateMessage();
+ String getRecoverTemplateMessage();
+ String getSuccessTemplateMessage();
+ }
+
+ class ProcessRecordSetStrategy implements ProcessStrategy {
+
+ static final String PROVENANCE_EVENT_DETAILS_ON_RECORDSET_FAILURE = "Publish failed after %d successfully published records.";
+ static final String PROVENANCE_EVENT_DETAILS_ON_RECORDSET_RECOVER = "Successfully finished publishing previously failed records. Total record count: %d";
+ static final String PROVENANCE_EVENT_DETAILS_ON_RECORDSET_SUCCESS = "Successfully published all records. Total record count: %d";
+
+ @Override
+ public void process(ProcessContext context, FlowFile flowfile, InputStream in, String topic, AtomicInteger processedRecords, Long previousProcessFailedAt) throws IOException {
+ final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+
+ try (final RecordReader reader = readerFactory.createRecordReader(flowfile, in, logger)) {
+ final RecordSet recordSet = reader.createRecordSet();
+
+ final RecordSchema schema = writerFactory.getSchema(flowfile.getAttributes(), recordSet.getSchema());
+
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream(1024);
+
+ Record record;
+ while ((record = recordSet.next()) != null) {
+ if (previousProcessFailedAt != null && processedRecords.get() < previousProcessFailedAt) {
+ processedRecords.getAndIncrement();
+ continue;
+ }
+
+ baos.reset();
+
+ try (final RecordSetWriter writer = writerFactory.createWriter(logger, schema, baos, flowfile)) {
+ writer.write(record);
+ writer.flush();
+ }
+
+ final byte[] messageContent = baos.toByteArray();
+
+ publishMessage(context, flowfile, topic, messageContent);
+ processedRecords.getAndIncrement();
+ }
+ } catch (SchemaNotFoundException | MalformedRecordException e) {
+ throw new ProcessException("An error happened during creating components for serialization.", e);
+ }
+ }
+
+ @Override
+ public String getFailureTemplateMessage() {
+ return PROVENANCE_EVENT_DETAILS_ON_RECORDSET_FAILURE;
+ }
+
+ @Override
+ public String getRecoverTemplateMessage() {
+ return PROVENANCE_EVENT_DETAILS_ON_RECORDSET_RECOVER;
+ }
+
+ @Override
+ public String getSuccessTemplateMessage() {
+ return PROVENANCE_EVENT_DETAILS_ON_RECORDSET_SUCCESS;
+ }
+ }
+
+ class ProcessDemarcatedContentStrategy implements ProcessStrategy {
+
+ static final String PROVENANCE_EVENT_DETAILS_ON_SINGLE_MESSAGE_FAILURE = "Publish failed after %d successfully published messages.";
+ static final String PROVENANCE_EVENT_DETAILS_ON_SINGLE_MESSAGE_RECOVER = "Successfully finished publishing previously failed messages. Total message count: %d";
+ static final String PROVENANCE_EVENT_DETAILS_ON_SINGLE_MESSAGE_SUCCESS = "Successfully published all messages. Total message count: %d";
Review Comment:
It is not really clear for me what `SINGLE_MESSAGE` means here. `DEMARCATED_MESSAGE` or `DEMARCATED_CONTENT` might be more appropriate.
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/PublishMQTT.java:
##########
@@ -109,6 +113,15 @@ public class PublishMQTT extends AbstractMQTTProcessor implements MqttCallback {
.description("The Record Writer to use for serializing Records before publishing them as an MQTT Message.")
.build();
+ public static final PropertyDescriptor MESSAGE_DEMARCATOR = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(BASE_MESSAGE_DEMARCATOR)
+ .description("With this property, you have an option to publish multiple messages from a single FlowFile. "
+ + "This property allows you to provide a string (interpreted as UTF-8) to use for splitting "
+ + "the FlowFile content. This is an optional property ; if not provided, and if not defining a "
+ + "Record Reader/Writer, each message received will result in a single FlowFile. To enter special "
Review Comment:
Copy-paste error from `ConsumeMQTT`:
```suggestion
+ "Record Reader/Writer, each FlowFile will be published as a single message. To enter special "
```
--
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 a diff in pull request #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
Posted by GitBox <gi...@apache.org>.
nandorsoma commented on code in PR #6530:
URL: https://github.com/apache/nifi/pull/6530#discussion_r996491653
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/PublishMQTT.java:
##########
@@ -109,6 +113,15 @@ public class PublishMQTT extends AbstractMQTTProcessor implements MqttCallback {
.description("The Record Writer to use for serializing Records before publishing them as an MQTT Message.")
.build();
+ public static final PropertyDescriptor MESSAGE_DEMARCATOR = new PropertyDescriptor.Builder()
+ .fromPropertyDescriptor(BASE_MESSAGE_DEMARCATOR)
+ .description("With this property, you have an option to publish multiple messages from a single FlowFile. "
+ + "This property allows you to provide a string (interpreted as UTF-8) to use for splitting "
+ + "the FlowFile content. This is an optional property ; if not provided, and if not defining a "
+ + "Record Reader/Writer, each message received will result in a single FlowFile. To enter special "
Review Comment:
Oh, I tried to match this description to the processor, but seems like I missed this sentence. Thx.
--
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 a diff in pull request #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
Posted by GitBox <gi...@apache.org>.
nandorsoma commented on code in PR #6530:
URL: https://github.com/apache/nifi/pull/6530#discussion_r996492341
##########
nifi-nar-bundles/nifi-mqtt-bundle/nifi-mqtt-processors/src/main/java/org/apache/nifi/processors/mqtt/PublishMQTT.java:
##########
@@ -336,4 +315,110 @@ public void deliveryComplete(String token) {
logger.trace("Received 'delivery complete' message from broker. Token: [{}]", token);
}
+ interface ProcessStrategy {
+ void process(ProcessContext context, FlowFile flowfile, InputStream in, String topic, AtomicInteger processedRecords, Long previousProcessFailedAt) throws IOException;
+ String getFailureTemplateMessage();
+ String getRecoverTemplateMessage();
+ String getSuccessTemplateMessage();
+ }
+
+ class ProcessRecordSetStrategy implements ProcessStrategy {
+
+ static final String PROVENANCE_EVENT_DETAILS_ON_RECORDSET_FAILURE = "Publish failed after %d successfully published records.";
+ static final String PROVENANCE_EVENT_DETAILS_ON_RECORDSET_RECOVER = "Successfully finished publishing previously failed records. Total record count: %d";
+ static final String PROVENANCE_EVENT_DETAILS_ON_RECORDSET_SUCCESS = "Successfully published all records. Total record count: %d";
+
+ @Override
+ public void process(ProcessContext context, FlowFile flowfile, InputStream in, String topic, AtomicInteger processedRecords, Long previousProcessFailedAt) throws IOException {
+ final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+
+ try (final RecordReader reader = readerFactory.createRecordReader(flowfile, in, logger)) {
+ final RecordSet recordSet = reader.createRecordSet();
+
+ final RecordSchema schema = writerFactory.getSchema(flowfile.getAttributes(), recordSet.getSchema());
+
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream(1024);
+
+ Record record;
+ while ((record = recordSet.next()) != null) {
+ if (previousProcessFailedAt != null && processedRecords.get() < previousProcessFailedAt) {
+ processedRecords.getAndIncrement();
+ continue;
+ }
+
+ baos.reset();
+
+ try (final RecordSetWriter writer = writerFactory.createWriter(logger, schema, baos, flowfile)) {
+ writer.write(record);
+ writer.flush();
+ }
+
+ final byte[] messageContent = baos.toByteArray();
+
+ publishMessage(context, flowfile, topic, messageContent);
+ processedRecords.getAndIncrement();
+ }
+ } catch (SchemaNotFoundException | MalformedRecordException e) {
+ throw new ProcessException("An error happened during creating components for serialization.", e);
+ }
+ }
+
+ @Override
+ public String getFailureTemplateMessage() {
+ return PROVENANCE_EVENT_DETAILS_ON_RECORDSET_FAILURE;
+ }
+
+ @Override
+ public String getRecoverTemplateMessage() {
+ return PROVENANCE_EVENT_DETAILS_ON_RECORDSET_RECOVER;
+ }
+
+ @Override
+ public String getSuccessTemplateMessage() {
+ return PROVENANCE_EVENT_DETAILS_ON_RECORDSET_SUCCESS;
+ }
+ }
+
+ class ProcessDemarcatedContentStrategy implements ProcessStrategy {
+
+ static final String PROVENANCE_EVENT_DETAILS_ON_SINGLE_MESSAGE_FAILURE = "Publish failed after %d successfully published messages.";
+ static final String PROVENANCE_EVENT_DETAILS_ON_SINGLE_MESSAGE_RECOVER = "Successfully finished publishing previously failed messages. Total message count: %d";
+ static final String PROVENANCE_EVENT_DETAILS_ON_SINGLE_MESSAGE_SUCCESS = "Successfully published all messages. Total message count: %d";
Review Comment:
I'm not good in naming. Thx, good idea.
--
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 #6530: NIFI-10644 Add Message Demarcator-style processing in PublishMQTT
Posted by GitBox <gi...@apache.org>.
nandorsoma commented on PR #6530:
URL: https://github.com/apache/nifi/pull/6530#issuecomment-1280054862
Hey @turcsanyip! Thank you for your review, I've applied the requested changes!
--
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