bbende commented on code in PR #9138:
URL: https://github.com/apache/nifi/pull/9138#discussion_r1705416023
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ParameterContextResource.java:
##########
@@ -323,6 +356,204 @@ public Response updateParameterContext(
);
}
+ @POST
+ @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{contextId}/assets")
+ @Operation(
+ summary = "Creates a new Asset in the given Parameter Context",
+ responses = @ApiResponse(content = @Content(schema =
@Schema(implementation = AssetEntity.class))),
+ description = "This endpoint will create a new Asset in the given
Parameter Context. The Asset will be created with the given name and the
contents of the file that is uploaded. " +
+ "The Asset will be created in the given Parameter Context,
and will be available for use by any component that references the Parameter
Context.",
+ security = {
+ @SecurityRequirement(name = "Read -
/parameter-contexts/{parameterContextId}"),
+ @SecurityRequirement(name = "Write -
/parameter-contexts/{parameterContextId}"),
+ @SecurityRequirement(name = "Read - for every component that
is affected by the update"),
+ @SecurityRequirement(name = "Write - for every component that
is affected by the update"),
+ @SecurityRequirement(name = "Read - for every currently
inherited parameter context")
+ }
+ )
+ @ApiResponses(
+ value = {
+ @ApiResponse(responseCode = "400", description = "NiFi was unable
to complete the request because it was invalid. The request should not be
retried without modification."),
+ @ApiResponse(responseCode = "401", description = "Client could not
be authenticated."),
+ @ApiResponse(responseCode = "403", description = "Client is not
authorized to make this request."),
+ @ApiResponse(responseCode = "404", description = "The specified
resource could not be found."),
+ @ApiResponse(responseCode = "409", description = "The request was
valid but NiFi was not in the appropriate state to process it.")
+ }
+ )
+ public Response createAsset(
+ @PathParam("contextId") final String contextId,
+ @HeaderParam(FILENAME_HEADER) final String assetName,
+ @Parameter(description = "The contents of the asset.", required =
true) final InputStream assetContents) throws IOException {
+
+ // Validate input
+ if (StringUtils.isBlank(assetName)) {
+ throw new IllegalArgumentException(FILENAME_HEADER + " header is
required");
+ }
+ if (assetContents == null) {
+ throw new IllegalArgumentException("Asset contents must be
specified.");
+ }
+
+ // If clustered and not all nodes are connected, do not allow creating
an asset.
+ // Generally, we allow the flow to be modified when nodes are
disconnected, but we do not allow creating an asset because
+ // the cluster has no mechanism for synchronizing those assets after
the upload.
+ final ClusterCoordinator clusterCoordinator = getClusterCoordinator();
+ if (clusterCoordinator != null) {
+ final Set<NodeIdentifier> disconnectedNodes =
clusterCoordinator.getNodeIdentifiers(NodeConnectionState.CONNECTING,
NodeConnectionState.DISCONNECTED, NodeConnectionState.DISCONNECTING);
+ if (!disconnectedNodes.isEmpty()) {
+ throw new IllegalStateException("Cannot create an Asset
because the following %s nodes are not currently connected:
%s".formatted(disconnectedNodes.size(), disconnectedNodes));
+ }
+ }
+
+ // Get the context or throw ResourceNotFoundException
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+ final ParameterContextEntity contextEntity =
serviceFacade.getParameterContext(contextId, false, user);
+ final Set<AffectedComponentEntity> affectedComponents =
serviceFacade.getComponentsAffectedByParameterContextUpdate(Collections.singletonList(contextEntity.getComponent()));
+
+ // Authorize the request
+ serviceFacade.authorizeAccess(lookup -> {
+ // Verify READ and WRITE permissions for user, for the Parameter
Context itself
+ final ParameterContext parameterContext =
lookup.getParameterContext(contextId);
+ parameterContext.authorize(authorizer, RequestAction.READ, user);
+ parameterContext.authorize(authorizer, RequestAction.WRITE, user);
+
+ // Verify READ and WRITE permissions for user, for every component
that is affected
+ affectedComponents.forEach(component ->
parameterUpdateManager.authorizeAffectedComponent(component, lookup, user,
true, true));
+ });
+
+ // If we need to replicate the request, we do so using the Upload
Request Replicator, rather than the typical replicate() method.
+ // This is because Upload Request Replication works differently in
that it needs to be able to replicate the InputStream multiple times,
+ // so it must create a file on disk to do so and then use the file's
content to replicate the request. It also bypasses the two-phase
+ // commit process that is used for other requests because doing so
would result in uploading the file twice to each node or providing a
+ // different request for each of the two phases.
+
+ final long startTime = System.currentTimeMillis();
+ final InputStream maxLengthInputStream = new
MaxLengthInputStream(assetContents, (long) DataUnit.GB.toB(1));
+
+ final AssetEntity assetEntity;
+ if (isReplicateRequest()) {
+ final UploadRequest<AssetEntity> uploadRequest = new
UploadRequest.Builder<AssetEntity>()
+ .user(NiFiUserUtils.getNiFiUser())
+ .filename(assetName)
+ .identifier(UUID.randomUUID().toString())
+ .contents(maxLengthInputStream)
+ .header(FILENAME_HEADER, assetName)
+ .header(CONTENT_TYPE_HEADER, UPLOAD_CONTENT_TYPE)
+ .exampleRequestUri(getAbsolutePath())
+ .responseClass(AssetEntity.class)
+ .successfulResponseStatus(HttpResponseStatus.OK.getCode())
+ .build();
+ assetEntity = uploadRequestReplicator.upload(uploadRequest);
+ } else {
+ final String existingContextId = contextEntity.getId();
+ final String sanitizedAssetName =
FileUtils.sanitizeFilename(assetName);
Review Comment:
That probably does make sense, will play around with that.
--
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]
