naddym commented on a change in pull request #5254:
URL: https://github.com/apache/nifi/pull/5254#discussion_r680126091
##########
File path:
nifi-external/nifi-kafka-connect/nifi-kafka-connector/src/main/java/org/apache/nifi/kafka/connect/StatelessNiFiSourceTask.java
##########
@@ -174,10 +174,17 @@ public void start(final Map<String, String> properties) {
partitionMap = clusterStatePartitionMap;
}
- for (final FlowFile flowFile : outputFlowFiles) {
- final byte[] contents = triggerResult.readContent(flowFile);
- final SourceRecord sourceRecord = createSourceRecord(flowFile,
contents, componentState, partitionMap);
- sourceRecords.add(sourceRecord);
+ try {
+ for (final FlowFile flowFile : outputFlowFiles) {
+ final byte[] contents = triggerResult.readContent(flowFile);
+ final SourceRecord sourceRecord = createSourceRecord(flowFile,
contents, componentState, partitionMap);
+ sourceRecords.add(sourceRecord);
+ }
+ } catch (final Exception e) {
+ logger.error("Failed to obtain contents of Output FlowFiles in
order to form Kafka Record", e);
+ triggerResult.abort(e);
+ failureYieldExpiration = System.currentTimeMillis() + 1000L; //
delay next execution for 1 second to avoid constnatly failing and utilization
huge amounts of resources
Review comment:
Just to heads up, small typo in comment, should be `constantly` instead
of `constnatly`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
