exceptionfactory commented on a change in pull request #5020:
URL: https://github.com/apache/nifi/pull/5020#discussion_r618601572
##########
File path:
nifi-nar-bundles/nifi-kudu-bundle/nifi-kudu-processors/src/main/java/org/apache/nifi/processors/kudu/PutKudu.java
##########
@@ -342,38 +346,73 @@ public void onTrigger(final ProcessContext context, final
ProcessSession session
final KerberosUser user = getKerberosUser();
if (user == null) {
- executeOnKuduClient(kuduClient -> trigger(context, session,
flowFiles, kuduClient));
+ executeOnKuduClient(kuduClient -> processFlowFiles(context,
session, flowFiles, kuduClient));
return;
}
final PrivilegedExceptionAction<Void> privilegedAction = () -> {
- executeOnKuduClient(kuduClient -> trigger(context, session,
flowFiles, kuduClient));
+ executeOnKuduClient(kuduClient -> processFlowFiles(context,
session, flowFiles, kuduClient));
return null;
};
final KerberosAction<Void> action = new KerberosAction<>(user,
privilegedAction, getLogger());
action.execute();
}
- private void trigger(final ProcessContext context, final ProcessSession
session, final List<FlowFile> flowFiles, KuduClient kuduClient) throws
ProcessException {
- final RecordReaderFactory recordReaderFactory =
context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
+ private void processFlowFiles(final ProcessContext context, final
ProcessSession session, final List<FlowFile> flowFiles, final KuduClient
kuduClient) {
+ final Map<FlowFile, Integer> processedRecords = new HashMap<>();
+ final Map<FlowFile, Object> flowFileFailures = new HashMap<>();
+ final Map<Operation, FlowFile> operationFlowFileMap = new HashMap<>();
+ final List<RowError> pendingRowErrors = new ArrayList<>();
final KuduSession kuduSession = createKuduSession(kuduClient);
+ try {
+ processFlowFiles(flowFiles,
+ processedRecords,
+ flowFileFailures,
+ operationFlowFileMap,
+ pendingRowErrors,
+ session,
+ context,
+ kuduClient,
+ kuduSession);
+ } finally {
+ try {
+ flushKuduSession(kuduSession, true, pendingRowErrors);
+ } catch (final KuduException|RuntimeException e) {
+ getLogger().error("KuduSession.close() Failed", e);
+ }
+ }
- final Map<FlowFile, Integer> numRecords = new HashMap<>();
- final Map<FlowFile, Object> flowFileFailures = new HashMap<>();
- final Map<Operation, FlowFile> operationFlowFileMap = new HashMap<>();
+ if (isRollbackOnFailure() && (!pendingRowErrors.isEmpty() ||
!flowFileFailures.isEmpty())) {
+ logFailures(pendingRowErrors, operationFlowFileMap);
+ session.rollback();
+ context.yield();
+ } else {
+ transferFlowFiles(flowFiles, processedRecords, flowFileFailures,
operationFlowFileMap, pendingRowErrors, session);
+ }
+ }
- int numBuffered = 0;
+ private void processFlowFiles(final List<FlowFile> flowFiles,
Review comment:
Updated to processRecords, thanks!
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
