mtien-apache commented on a change in pull request #4846:
URL: https://github.com/apache/nifi/pull/4846#discussion_r583918059
##########
File path:
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
##########
@@ -4173,6 +4187,205 @@ private void sanitizeRegistryInfo(final
VersionedProcessGroup versionedProcessGr
}
}
+ /**
+ * Uploads the specified versioned flow definition and adds it to a new
process group.
+ *
+ * @param httpServletRequest request
+ * @param in The flow definition stream
+ * @return A processGroupEntity
+ * @throws IOException if there is an error during deserialization of the
InputStream
+ */
+ @POST
+ @Consumes(MediaType.MULTIPART_FORM_DATA)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{id}/process-groups/upload")
+ @ApiOperation(
+ value = "Uploads a versioned flow definition and creates a process
group",
+ response = ProcessGroupEntity.class,
+ authorizations = {
+ @Authorization(value = "Write - /process-groups/{uuid}")
+ }
+ )
+ @ApiResponses(
+ value = {
+ @ApiResponse(code = 400, message = "NiFi was unable to
complete the request because it was invalid. The request should not be retried
without modification."),
+ @ApiResponse(code = 401, message = "Client could not be
authenticated."),
+ @ApiResponse(code = 403, message = "Client is not
authorized to make this request."),
+ @ApiResponse(code = 404, message = "The specified resource
could not be found."),
+ @ApiResponse(code = 409, message = "The request was valid
but NiFi was not in the appropriate state to process it. Retrying the same
request later may be successful.")
+ }
+ )
+ public Response uploadProcessGroup(
+ @Context final HttpServletRequest httpServletRequest,
+ @ApiParam(
+ value = "The process group id.",
+ required = true
+ )
+ @PathParam("id") final String groupId,
+ @ApiParam(
+ value = "The process group name.",
+ required = true
+ )
+ @FormDataParam("groupName") final String groupName,
+ @ApiParam(
+ value = "The process group X position.",
+ required = true
+ )
+ @FormDataParam("position-x") final Double positionX,
+ @ApiParam(
+ value = "The process group Y position.",
+ required = true
+ )
+ @FormDataParam("position-y") final Double positionY,
+ @ApiParam(
+ value = "The client id.",
+ required = true
+ )
+ @FormDataParam("clientId") final String clientId,
+ @ApiParam(
+ value = "Acknowledges that this node is disconnected to
allow for mutable requests to proceed.",
+ required = false
+ )
+ @FormDataParam(DISCONNECTED_NODE_ACKNOWLEDGED)
@DefaultValue("false") final Boolean disconnectedNodeAcknowledged,
+ @FormDataParam("file") final InputStream in) throws IOException {
+
+ // ensure the group name is specified
+ if (StringUtils.isBlank(groupName)) {
+ throw new IllegalArgumentException("The process group name is
required.");
+ }
+
+ if (StringUtils.isBlank(groupId)) {
+ throw new IllegalArgumentException("The parent process group id is
required");
+ }
+
+ if (positionX == null) {
+ throw new IllegalArgumentException("The x coordinate of the
proposed position must be specified.");
+ }
+
+ if (positionY == null) {
+ throw new IllegalArgumentException("The y coordinate of the
proposed position must be specified.");
+ }
+
+ if (StringUtils.isBlank(clientId)) {
+ throw new IllegalArgumentException("The client id must be
specified");
+ }
+
+ // get the contents of the InputStream as a String
+ String stringContent;
+ if (in != null) {
+ try {
+ stringContent = IOUtils.toString(in, StandardCharsets.UTF_8);
+ } catch (IOException e) {
+ throw new IOException("Unable to read the InputStream", e);
+ }
+ } else {
+ logger.warn("The InputStream is null");
+ throw new NullPointerException("The InputStream is null");
+ }
+
+ // deserialize content to a VersionedFlowSnapshot
+ VersionedFlowSnapshot deserializedSnapshot;
+
+ if (stringContent.length() > 0) {
+ try {
+ deserializedSnapshot = MAPPER.readValue(stringContent,
VersionedFlowSnapshot.class);
+ } catch (JsonParseException jpe) {
+ logger.warn("Parsing uploaded JSON failed", jpe);
+ return
Response.status(Response.Status.OK).entity(INVALID_JSON_RESPONSE).type("application/json").build();
+ } catch (IOException e) {
+ logger.warn("Deserialization of uploaded JSON failed", e);
+ throw new IOException("Deserialization of uploaded JSON
failed", e);
+ }
+ } else {
+ logger.warn("The uploaded file was empty");
+ throw new IOException("The uploaded file was empty.");
+ }
+
+ // create a new ProcessGroupEntity
+ final ProcessGroupEntity newProcessGroupEntity =
createProcessGroupEntity(groupId, groupName, positionX, positionY,
deserializedSnapshot);
+
+ // replicate the request or call serviceFacade.updateProcessGroup
+ if (isReplicateRequest()) {
+ return replicate(HttpMethod.POST, newProcessGroupEntity);
+ } else if (isDisconnectedFromCluster()) {
+
verifyDisconnectedNodeModification(newProcessGroupEntity.isDisconnectedNodeAcknowledged());
+ }
+
+ return withWriteLock(
+ serviceFacade,
+ newProcessGroupEntity,
+ lookup -> {
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+ final Authorizable processGroup =
lookup.getProcessGroup(groupId).getAuthorizable();
+ processGroup.authorize(authorizer, RequestAction.WRITE,
user);
+
+ // if request specifies a Parameter Context, need to
authorize that user has READ policy for the Parameter Context.
+ final ParameterContextReferenceEntity
referencedParamContext =
newProcessGroupEntity.getComponent().getParameterContext();
+ if (referencedParamContext != null &&
referencedParamContext.getId() != null) {
+
lookup.getParameterContext(referencedParamContext.getId()).authorize(authorizer,
RequestAction.READ, user);
+ }
+
+ // if any of the components is a Restricted Component,
then we must authorize the user
+ // for write access to the RestrictedComponents resource
+ final VersionedFlowSnapshot versionedFlowSnapshot =
newProcessGroupEntity.getVersionedFlowSnapshot();
+ if (versionedFlowSnapshot != null) {
+ final Set<ConfigurableComponent> restrictedComponents
=
FlowRegistryUtils.getRestrictedComponents(versionedFlowSnapshot.getFlowContents(),
serviceFacade);
+ restrictedComponents.forEach(restrictedComponent -> {
+ final ComponentAuthorizable
restrictedComponentAuthorizable =
lookup.getConfigurableComponent(restrictedComponent);
+ authorizeRestrictions(authorizer,
restrictedComponentAuthorizable);
+ });
+
+ final Map<String, VersionedParameterContext>
parameterContexts = versionedFlowSnapshot.getParameterContexts();
+ if (parameterContexts != null) {
+ parameterContexts.values().forEach(context ->
AuthorizeParameterReference.authorizeParameterContextAddition(context,
serviceFacade, authorizer, lookup, user));
+ }
+ }
+ },
+ () -> {
+ final VersionedFlowSnapshot versionedFlowSnapshot =
newProcessGroupEntity.getVersionedFlowSnapshot();
+ if (versionedFlowSnapshot != null) {
+
serviceFacade.verifyComponentTypes(versionedFlowSnapshot.getFlowContents());
+ }
+ },
+ processGroupEntity -> {
+ final ProcessGroupDTO processGroup =
processGroupEntity.getComponent();
+
+ // set the processor id as appropriate
+ processGroup.setId(generateUuid());
+
+ // get the versioned flow
+ final VersionedFlowSnapshot flowSnapshot =
processGroupEntity.getVersionedFlowSnapshot();
+
+ // create the process group contents
+ final Revision revision = new Revision((long) 0, clientId,
processGroup.getId());
+
+ ProcessGroupEntity entity =
serviceFacade.createProcessGroup(revision, groupId, processGroup);
+
+ if (flowSnapshot != null) {
+ final RevisionDTO revisionDto = entity.getRevision();
+ final String newGroupId =
entity.getComponent().getId();
+ final Revision newGroupRevision = new
Revision(revisionDto.getVersion(), revisionDto.getClientId(), newGroupId);
+
+ // We don't want the Process Group's position to be
updated because we want to keep the position where the user
+ // placed the Process Group. We do not want to use the
name of the Process Group that is in the Flow Contents.
+ // To accomplish this, we call
updateProcessGroupContents() passing 'false' for the updateSettings flag, set
+ // the Process Group name, and null out the position.
+ flowSnapshot.getFlowContents().setPosition(null);
+ flowSnapshot.getFlowContents().setName(groupName);
+
+ entity =
serviceFacade.updateProcessGroupContents(newGroupRevision, newGroupId, null,
flowSnapshot,
+ getIdGenerationSeed().orElse(null), false,
false, true);
+ }
+
+ populateRemainingProcessGroupEntityContent(entity);
+
+ // generate a 201 created response
+ String uri = entity.getUri();
+ return generateCreatedResponse(URI.create(uri),
entity).build();
+ }
+ );
Review comment:
I believe I can partially refactor this block. Will make the change.
----------------------------------------------------------------
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:
[email protected]
