You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "turcsanyip (via GitHub)" <gi...@apache.org> on 2026/01/06 17:51:36 UTC
[PR] NIFI-15431 Fixed error handling in PutSmbFile to allow batch processi… [nifi]
turcsanyip opened a new pull request, #10734:
URL: https://github.com/apache/nifi/pull/10734
…ng to continue
# Summary
[NIFI-15431](https://issues.apache.org/jira/browse/NIFI-15431)
# Tracking
Please complete the following tracking steps prior to pull request creation.
### Issue Tracking
- [ ] [Apache NiFi Jira](https://issues.apache.org/jira/browse/NIFI) issue created
### Pull Request Tracking
- [ ] Pull Request title starts with Apache NiFi Jira issue number, such as `NIFI-00000`
- [ ] Pull Request commit message starts with Apache NiFi Jira issue number, as such `NIFI-00000`
- [ ] Pull request contains [commits signed](https://docs.github.com/en/authentication/managing-commit-signature-verification/signing-commits) with a registered key indicating `Verified` status
### Pull Request Formatting
- [ ] Pull Request based on current revision of the `main` branch
- [ ] 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
- [ ] Build completed using `./mvnw clean install -P contrib-check`
- [ ] JDK 21
- [ ] JDK 25
### 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
Re: [PR] NIFI-15431 Fixed error handling in PutSmbFile to allow batch processi… [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on PR #10734:
URL: https://github.com/apache/nifi/pull/10734#issuecomment-3729042474
Forgot the magic words to close the PR. This has been merged via https://github.com/apache/nifi/commit/3a94d2688365a19695b1f0e7eeedae291c8801ed.
--
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-15431 Fixed error handling in PutSmbFile to allow batch processi… [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 commented on code in PR #10734:
URL: https://github.com/apache/nifi/pull/10734#discussion_r2675775350
##########
nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java:
##########
@@ -316,106 +316,119 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
DiskShare share = (DiskShare) smbSession.connectShare(shareName)) {
for (FlowFile flowFile : flowFiles) {
- final long processingStartTime = System.nanoTime();
+ try {
+ final long processingStartTime = System.nanoTime();
- final String destinationDirectory = context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue();
- final String destinationFilename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
+ final String destinationDirectory = context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue();
+ final String destinationFilename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
- String destinationFullPath;
+ String destinationFullPath;
- // build destination path for the flowfile
- if (destinationDirectory == null || destinationDirectory.isBlank()) {
- destinationFullPath = destinationFilename;
- } else {
- destinationFullPath = new java.io.File(destinationDirectory, destinationFilename).getPath();
- }
-
- // handle missing directory
- final String destinationFileParentDirectory = new java.io.File(destinationFullPath).getParent();
- final Boolean createMissingDirectories = context.getProperty(CREATE_DIRS).asBoolean();
- if (!createMissingDirectories && !share.folderExists(destinationFileParentDirectory)) {
- flowFile = session.penalize(flowFile);
- logger.warn("Penalizing {} and routing to failure as configured because the destination directory ({}) doesn't exist", flowFile, destinationFileParentDirectory);
- session.transfer(flowFile, REL_FAILURE);
- continue;
- } else if (!share.folderExists(destinationFileParentDirectory)) {
- createMissingDirectoriesRecursevly(logger, share, destinationFileParentDirectory);
- }
+ // build destination path for the flowfile
+ if (destinationDirectory == null || destinationDirectory.isBlank()) {
+ destinationFullPath = destinationFilename;
+ } else {
+ destinationFullPath = new java.io.File(destinationDirectory, destinationFilename).getPath();
+ }
- // handle conflict resolution
- final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
- if (share.fileExists(destinationFullPath)) {
- if (conflictResolution.equals(IGNORE_RESOLUTION)) {
- session.transfer(flowFile, REL_SUCCESS);
- logger.info("Transferring {} to success as configured because file with same name already exists", flowFile);
- continue;
- } else if (conflictResolution.equals(FAIL_RESOLUTION)) {
+ // handle missing directory
+ final String destinationFileParentDirectory = new java.io.File(destinationFullPath).getParent();
+ final Boolean createMissingDirectories = context.getProperty(CREATE_DIRS).asBoolean();
+ if (!createMissingDirectories && !share.folderExists(destinationFileParentDirectory)) {
+ logger.warn("Penalizing {} and routing to failure as configured because the destination directory ({}) doesn't exist", flowFile, destinationFileParentDirectory);
flowFile = session.penalize(flowFile);
- logger.warn("Penalizing {} and routing to failure as configured because file with the same name already exists", flowFile);
session.transfer(flowFile, REL_FAILURE);
continue;
+ } else if (!share.folderExists(destinationFileParentDirectory)) {
+ try {
+ createMissingDirectoriesRecursively(logger, share, destinationFileParentDirectory);
+ } catch (Exception e) {
+ logger.error("Penalizing {} and routing to failure because failed to create missing destination directories ({})", flowFile, destinationFileParentDirectory);
Review Comment:
Do we want to include the exception in the log message?
##########
nifi-extension-bundles/nifi-smb-bundle/nifi-smb-processors/src/main/java/org/apache/nifi/processors/smb/PutSmbFile.java:
##########
@@ -316,106 +316,119 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
DiskShare share = (DiskShare) smbSession.connectShare(shareName)) {
for (FlowFile flowFile : flowFiles) {
- final long processingStartTime = System.nanoTime();
+ try {
+ final long processingStartTime = System.nanoTime();
- final String destinationDirectory = context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue();
- final String destinationFilename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
+ final String destinationDirectory = context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue();
+ final String destinationFilename = flowFile.getAttribute(CoreAttributes.FILENAME.key());
- String destinationFullPath;
+ String destinationFullPath;
- // build destination path for the flowfile
- if (destinationDirectory == null || destinationDirectory.isBlank()) {
- destinationFullPath = destinationFilename;
- } else {
- destinationFullPath = new java.io.File(destinationDirectory, destinationFilename).getPath();
- }
-
- // handle missing directory
- final String destinationFileParentDirectory = new java.io.File(destinationFullPath).getParent();
- final Boolean createMissingDirectories = context.getProperty(CREATE_DIRS).asBoolean();
- if (!createMissingDirectories && !share.folderExists(destinationFileParentDirectory)) {
- flowFile = session.penalize(flowFile);
- logger.warn("Penalizing {} and routing to failure as configured because the destination directory ({}) doesn't exist", flowFile, destinationFileParentDirectory);
- session.transfer(flowFile, REL_FAILURE);
- continue;
- } else if (!share.folderExists(destinationFileParentDirectory)) {
- createMissingDirectoriesRecursevly(logger, share, destinationFileParentDirectory);
- }
+ // build destination path for the flowfile
+ if (destinationDirectory == null || destinationDirectory.isBlank()) {
+ destinationFullPath = destinationFilename;
+ } else {
+ destinationFullPath = new java.io.File(destinationDirectory, destinationFilename).getPath();
+ }
- // handle conflict resolution
- final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
- if (share.fileExists(destinationFullPath)) {
- if (conflictResolution.equals(IGNORE_RESOLUTION)) {
- session.transfer(flowFile, REL_SUCCESS);
- logger.info("Transferring {} to success as configured because file with same name already exists", flowFile);
- continue;
- } else if (conflictResolution.equals(FAIL_RESOLUTION)) {
+ // handle missing directory
+ final String destinationFileParentDirectory = new java.io.File(destinationFullPath).getParent();
+ final Boolean createMissingDirectories = context.getProperty(CREATE_DIRS).asBoolean();
+ if (!createMissingDirectories && !share.folderExists(destinationFileParentDirectory)) {
+ logger.warn("Penalizing {} and routing to failure as configured because the destination directory ({}) doesn't exist", flowFile, destinationFileParentDirectory);
flowFile = session.penalize(flowFile);
- logger.warn("Penalizing {} and routing to failure as configured because file with the same name already exists", flowFile);
session.transfer(flowFile, REL_FAILURE);
continue;
+ } else if (!share.folderExists(destinationFileParentDirectory)) {
+ try {
+ createMissingDirectoriesRecursively(logger, share, destinationFileParentDirectory);
+ } catch (Exception e) {
+ logger.error("Penalizing {} and routing to failure because failed to create missing destination directories ({})", flowFile, destinationFileParentDirectory);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ continue;
+ }
}
- }
- // handle temporary suffix
- final String renameSuffixValue = context.getProperty(RENAME_SUFFIX).getValue();
- final boolean renameSuffix = renameSuffixValue != null && !renameSuffixValue.isBlank();
- StringBuilder finalDestinationFullPath = new StringBuilder(destinationFullPath);
- if (renameSuffix) {
- finalDestinationFullPath.append(renameSuffixValue);
- }
+ // handle conflict resolution
+ final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
+ if (share.fileExists(destinationFullPath)) {
+ if (conflictResolution.equals(IGNORE_RESOLUTION)) {
+ logger.info("Transferring {} to success as configured because file with same name already exists", flowFile);
+ session.transfer(flowFile, REL_SUCCESS);
+ continue;
+ } else if (conflictResolution.equals(FAIL_RESOLUTION)) {
+ logger.warn("Penalizing {} and routing to failure as configured because file with the same name already exists", flowFile);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ continue;
+ }
+ }
- // handle the transfer
- try (
- File shareDestinationFile = share.openFile(
- finalDestinationFullPath.toString(),
- EnumSet.of(AccessMask.GENERIC_WRITE),
- EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL),
- sharedAccess,
- SMB2CreateDisposition.FILE_OVERWRITE_IF,
- EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH));
- OutputStream shareDestinationFileOutputStream = shareDestinationFile.getOutputStream()) {
- session.exportTo(flowFile, shareDestinationFileOutputStream);
- } catch (Exception e) {
- flowFile = session.penalize(flowFile);
- session.transfer(flowFile, REL_FAILURE);
- logger.error("Cannot transfer the file. Penalizing {} and routing to 'failure'", flowFile, e);
- continue;
- }
+ // handle temporary suffix
+ final String renameSuffixValue = context.getProperty(RENAME_SUFFIX).getValue();
+ final boolean renameSuffix = renameSuffixValue != null && !renameSuffixValue.isBlank();
+ StringBuilder finalDestinationFullPath = new StringBuilder(destinationFullPath);
+ if (renameSuffix) {
+ finalDestinationFullPath.append(renameSuffixValue);
+ }
- // handle the rename
- if (renameSuffix) {
- try (DiskEntry fileDiskEntry = share.open(
- finalDestinationFullPath.toString(),
- EnumSet.of(AccessMask.DELETE, AccessMask.GENERIC_WRITE),
- EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL),
- sharedAccess,
- SMB2CreateDisposition.FILE_OPEN,
- EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH))) {
-
- // normalize path slashes for the network share
- destinationFullPath = destinationFullPath.replace("/", "\\");
-
- // rename the file on the share and replace it in case it exists
- fileDiskEntry.rename(destinationFullPath, true);
+ // handle the transfer
+ try (
+ File shareDestinationFile = share.openFile(
+ finalDestinationFullPath.toString(),
+ EnumSet.of(AccessMask.GENERIC_WRITE),
+ EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL),
+ sharedAccess,
+ SMB2CreateDisposition.FILE_OVERWRITE_IF,
+ EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH));
+ OutputStream shareDestinationFileOutputStream = shareDestinationFile.getOutputStream()) {
+ session.exportTo(flowFile, shareDestinationFileOutputStream);
} catch (Exception e) {
+ logger.error("Cannot transfer the file. Penalizing {} and routing to 'failure'", flowFile, e);
flowFile = session.penalize(flowFile);
session.transfer(flowFile, REL_FAILURE);
- logger.error("Cannot rename the file. Penalizing {} and routing to 'failure'", flowFile, e);
continue;
}
- }
- // handle the success
- final URI provenanceUri = new URI("smb", hostname, "/" + destinationFullPath.replace('\\', '/'), null);
- final long processingTimeInNano = System.nanoTime() - processingStartTime;
- final long processingTimeInMilli = TimeUnit.MILLISECONDS.convert(processingTimeInNano, TimeUnit.NANOSECONDS);
- session.getProvenanceReporter().send(flowFile, provenanceUri.toString(), processingTimeInMilli);
- session.transfer(flowFile, REL_SUCCESS);
+ // handle the rename
+ if (renameSuffix) {
+ try (DiskEntry fileDiskEntry = share.open(
+ finalDestinationFullPath.toString(),
+ EnumSet.of(AccessMask.DELETE, AccessMask.GENERIC_WRITE),
+ EnumSet.of(FileAttributes.FILE_ATTRIBUTE_NORMAL),
+ sharedAccess,
+ SMB2CreateDisposition.FILE_OPEN,
+ EnumSet.of(SMB2CreateOptions.FILE_WRITE_THROUGH))) {
+
+ // normalize path slashes for the network share
+ destinationFullPath = destinationFullPath.replace("/", "\\");
+
+ // rename the file on the share and replace it in case it exists
+ fileDiskEntry.rename(destinationFullPath, true);
+ } catch (Exception e) {
+ logger.error("Cannot rename the file. Penalizing {} and routing to 'failure'", flowFile, e);
+ flowFile = session.penalize(flowFile);
+ session.transfer(flowFile, REL_FAILURE);
+ continue;
+ }
+ }
+
+ // handle the success
+ final URI provenanceUri = new URI("smb", hostname, "/" + destinationFullPath.replace('\\', '/'), null);
+ final long processingTimeInNano = System.nanoTime() - processingStartTime;
+ final long processingTimeInMilli = TimeUnit.MILLISECONDS.convert(processingTimeInNano, TimeUnit.NANOSECONDS);
+ session.getProvenanceReporter().send(flowFile, provenanceUri.toString(), processingTimeInMilli);
+ session.transfer(flowFile, REL_SUCCESS);
+ } catch (Exception e) {
+ logger.error("Error processing flowfile", e);
Review Comment:
Do we want to include the flowfile?
```suggestion
logger.error("Error processing flowfile", flowFile, 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-15431 Fixed error handling in PutSmbFile to allow batch processi… [nifi]
Posted by "pvillard31 (via GitHub)" <gi...@apache.org>.
pvillard31 closed pull request #10734: NIFI-15431 Fixed error handling in PutSmbFile to allow batch processi…
URL: https://github.com/apache/nifi/pull/10734
--
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