@Test public void testCommitOutputPortTransaction() throws Exception { final HttpServletRequest req = createCommonHttpServletRequest(); final DataTransferResource resource = getDataTransferResource(); final ServletContext context = null; final InputStream inputStream = null; final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.commitOutputPortTransaction(ResponseCode.CONFIRM_TRANSACTION.getCode(), "client-checksum", "port-id", transactionId, req, context, inputStream); transactionManager.cancelTransaction(transactionId); TransactionResultEntity resultEntity = (TransactionResultEntity) response.getEntity(); assertEquals(200, response.getStatus()); assertEquals(ResponseCode.CONFIRM_TRANSACTION.getCode(), resultEntity.getResponseCode()); }
@DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response commitOutputPortTransaction( @ApiParam( value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).", required = true ) @QueryParam(RESPONSE_CODE) Integer responseCode, @ApiParam( value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.", required = true ) @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum, @ApiParam( value = "The output port id.", required = true ) @PathParam("portId") String portId, @ApiParam( value = "The transaction id.", required = true ) @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.OutputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("commitOutputPortTransaction request: portId={}, transactionId={}", portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final TransactionResultEntity entity = new TransactionResultEntity(); try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); String inputErrMessage = null; if (responseCode == null) { inputErrMessage = "responseCode is required."; } else if (ResponseCode.CONFIRM_TRANSACTION.getCode() != responseCode && ResponseCode.CANCEL_TRANSACTION.getCode() != responseCode) { inputErrMessage = "responseCode " + responseCode + " is invalid. "; } if (inputErrMessage != null) { entity.setMessage(inputErrMessage); entity.setResponseCode(ResponseCode.ABORT.getCode()); return Response.status(Response.Status.BAD_REQUEST).entity(entity).build(); } if (ResponseCode.CANCEL_TRANSACTION.getCode() == responseCode) { return cancelTransaction(transactionId, entity); } int flowFileSent = serverProtocol.commitTransferTransaction(peer, checksum); entity.setResponseCode(ResponseCode.CONFIRM_TRANSACTION.getCode()); entity.setFlowFileSent(flowFileSent); } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (Exception e) { HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); logger.error("Failed to process the request", e); if (ResponseCode.BAD_CHECKSUM.equals(commsSession.getResponseCode())) { entity.setResponseCode(commsSession.getResponseCode().getCode()); entity.setMessage(e.getMessage()); Response.ResponseBuilder builder = Response.status(Response.Status.BAD_REQUEST).entity(entity); return clusterContext(noCache(builder)).build(); } return responseCreator.unexpectedErrorResponse(portId, transactionId, e); } return clusterContext(noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager))).build(); }
DataTransferResource extends ApplicationResource { @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response commitOutputPortTransaction( @ApiParam( value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).", required = true ) @QueryParam(RESPONSE_CODE) Integer responseCode, @ApiParam( value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.", required = true ) @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum, @ApiParam( value = "The output port id.", required = true ) @PathParam("portId") String portId, @ApiParam( value = "The transaction id.", required = true ) @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.OutputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("commitOutputPortTransaction request: portId={}, transactionId={}", portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final TransactionResultEntity entity = new TransactionResultEntity(); try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); String inputErrMessage = null; if (responseCode == null) { inputErrMessage = "responseCode is required."; } else if (ResponseCode.CONFIRM_TRANSACTION.getCode() != responseCode && ResponseCode.CANCEL_TRANSACTION.getCode() != responseCode) { inputErrMessage = "responseCode " + responseCode + " is invalid. "; } if (inputErrMessage != null) { entity.setMessage(inputErrMessage); entity.setResponseCode(ResponseCode.ABORT.getCode()); return Response.status(Response.Status.BAD_REQUEST).entity(entity).build(); } if (ResponseCode.CANCEL_TRANSACTION.getCode() == responseCode) { return cancelTransaction(transactionId, entity); } int flowFileSent = serverProtocol.commitTransferTransaction(peer, checksum); entity.setResponseCode(ResponseCode.CONFIRM_TRANSACTION.getCode()); entity.setFlowFileSent(flowFileSent); } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (Exception e) { HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); logger.error("Failed to process the request", e); if (ResponseCode.BAD_CHECKSUM.equals(commsSession.getResponseCode())) { entity.setResponseCode(commsSession.getResponseCode().getCode()); entity.setMessage(e.getMessage()); Response.ResponseBuilder builder = Response.status(Response.Status.BAD_REQUEST).entity(entity); return clusterContext(noCache(builder)).build(); } return responseCreator.unexpectedErrorResponse(portId, transactionId, e); } return clusterContext(noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager))).build(); } }
DataTransferResource extends ApplicationResource { @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response commitOutputPortTransaction( @ApiParam( value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).", required = true ) @QueryParam(RESPONSE_CODE) Integer responseCode, @ApiParam( value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.", required = true ) @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum, @ApiParam( value = "The output port id.", required = true ) @PathParam("portId") String portId, @ApiParam( value = "The transaction id.", required = true ) @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.OutputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("commitOutputPortTransaction request: portId={}, transactionId={}", portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final TransactionResultEntity entity = new TransactionResultEntity(); try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); String inputErrMessage = null; if (responseCode == null) { inputErrMessage = "responseCode is required."; } else if (ResponseCode.CONFIRM_TRANSACTION.getCode() != responseCode && ResponseCode.CANCEL_TRANSACTION.getCode() != responseCode) { inputErrMessage = "responseCode " + responseCode + " is invalid. "; } if (inputErrMessage != null) { entity.setMessage(inputErrMessage); entity.setResponseCode(ResponseCode.ABORT.getCode()); return Response.status(Response.Status.BAD_REQUEST).entity(entity).build(); } if (ResponseCode.CANCEL_TRANSACTION.getCode() == responseCode) { return cancelTransaction(transactionId, entity); } int flowFileSent = serverProtocol.commitTransferTransaction(peer, checksum); entity.setResponseCode(ResponseCode.CONFIRM_TRANSACTION.getCode()); entity.setFlowFileSent(flowFileSent); } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (Exception e) { HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); logger.error("Failed to process the request", e); if (ResponseCode.BAD_CHECKSUM.equals(commsSession.getResponseCode())) { entity.setResponseCode(commsSession.getResponseCode().getCode()); entity.setMessage(e.getMessage()); Response.ResponseBuilder builder = Response.status(Response.Status.BAD_REQUEST).entity(entity); return clusterContext(noCache(builder)).build(); } return responseCreator.unexpectedErrorResponse(portId, transactionId, e); } return clusterContext(noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager))).build(); } DataTransferResource(final NiFiProperties nifiProperties); }
DataTransferResource extends ApplicationResource { @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response commitOutputPortTransaction( @ApiParam( value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).", required = true ) @QueryParam(RESPONSE_CODE) Integer responseCode, @ApiParam( value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.", required = true ) @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum, @ApiParam( value = "The output port id.", required = true ) @PathParam("portId") String portId, @ApiParam( value = "The transaction id.", required = true ) @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.OutputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("commitOutputPortTransaction request: portId={}, transactionId={}", portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final TransactionResultEntity entity = new TransactionResultEntity(); try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); String inputErrMessage = null; if (responseCode == null) { inputErrMessage = "responseCode is required."; } else if (ResponseCode.CONFIRM_TRANSACTION.getCode() != responseCode && ResponseCode.CANCEL_TRANSACTION.getCode() != responseCode) { inputErrMessage = "responseCode " + responseCode + " is invalid. "; } if (inputErrMessage != null) { entity.setMessage(inputErrMessage); entity.setResponseCode(ResponseCode.ABORT.getCode()); return Response.status(Response.Status.BAD_REQUEST).entity(entity).build(); } if (ResponseCode.CANCEL_TRANSACTION.getCode() == responseCode) { return cancelTransaction(transactionId, entity); } int flowFileSent = serverProtocol.commitTransferTransaction(peer, checksum); entity.setResponseCode(ResponseCode.CONFIRM_TRANSACTION.getCode()); entity.setFlowFileSent(flowFileSent); } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (Exception e) { HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); logger.error("Failed to process the request", e); if (ResponseCode.BAD_CHECKSUM.equals(commsSession.getResponseCode())) { entity.setResponseCode(commsSession.getResponseCode().getCode()); entity.setMessage(e.getMessage()); Response.ResponseBuilder builder = Response.status(Response.Status.BAD_REQUEST).entity(entity); return clusterContext(noCache(builder)).build(); } return responseCreator.unexpectedErrorResponse(portId, transactionId, e); } return clusterContext(noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager))).build(); } DataTransferResource(final NiFiProperties nifiProperties); @POST @Produces(MediaType.APPLICATION_JSON) @Path("{portType}/{portId}/transactions") @ApiOperation( value = "Create a transaction to the specified output port or input port", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/{component-type}/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response createPortTransaction(
            @ApiParam(
                    value = "The port type.",
                    required = true,
                    allowableValues = "input-ports, output-ports"
            )
            @PathParam("portType") String portType,
            @PathParam("portId") String portId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response receiveFlowFiles(
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitOutputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.",
                    required = true
            )
            @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum,
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitInputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are BAD_CHECKSUM(19), CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_OCTET_STREAM) @Path("output-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files from the output port", response = StreamingOutput.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 200, message = "There is no flow file to return."), @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response transferFlowFiles(
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @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.") } ) Response extendInputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response extendOutputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); Response extendPortTransactionTTL(
            String portType,
            String portId,
            String transactionId,
            HttpServletRequest req,
            HttpServletResponse res,
            ServletContext context,
            UriInfo uriInfo,
            InputStream inputStream); void setServiceFacade(NiFiServiceFacade serviceFacade); }
DataTransferResource extends ApplicationResource { @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) public Response commitOutputPortTransaction( @ApiParam( value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).", required = true ) @QueryParam(RESPONSE_CODE) Integer responseCode, @ApiParam( value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.", required = true ) @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum, @ApiParam( value = "The output port id.", required = true ) @PathParam("portId") String portId, @ApiParam( value = "The transaction id.", required = true ) @PathParam("transactionId") String transactionId, @Context HttpServletRequest req, @Context ServletContext context, InputStream inputStream) { serviceFacade.authorizeAccess(lookup -> { authorizeDataTransfer(lookup, ResourceType.OutputPort, portId); }); final ValidateRequestResult validationResult = validateResult(req, portId, transactionId); if (validationResult.errResponse != null) { return validationResult.errResponse; } logger.debug("commitOutputPortTransaction request: portId={}, transactionId={}", portId, transactionId); final int transportProtocolVersion = validationResult.transportProtocolVersion; final ByteArrayOutputStream out = new ByteArrayOutputStream(); final Peer peer = constructPeer(req, inputStream, out, portId, transactionId); final TransactionResultEntity entity = new TransactionResultEntity(); try { HttpFlowFileServerProtocol serverProtocol = initiateServerProtocol(req, peer, transportProtocolVersion); String inputErrMessage = null; if (responseCode == null) { inputErrMessage = "responseCode is required."; } else if (ResponseCode.CONFIRM_TRANSACTION.getCode() != responseCode && ResponseCode.CANCEL_TRANSACTION.getCode() != responseCode) { inputErrMessage = "responseCode " + responseCode + " is invalid. "; } if (inputErrMessage != null) { entity.setMessage(inputErrMessage); entity.setResponseCode(ResponseCode.ABORT.getCode()); return Response.status(Response.Status.BAD_REQUEST).entity(entity).build(); } if (ResponseCode.CANCEL_TRANSACTION.getCode() == responseCode) { return cancelTransaction(transactionId, entity); } int flowFileSent = serverProtocol.commitTransferTransaction(peer, checksum); entity.setResponseCode(ResponseCode.CONFIRM_TRANSACTION.getCode()); entity.setFlowFileSent(flowFileSent); } catch (HandshakeException e) { return responseCreator.handshakeExceptionResponse(e); } catch (Exception e) { HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession(); logger.error("Failed to process the request", e); if (ResponseCode.BAD_CHECKSUM.equals(commsSession.getResponseCode())) { entity.setResponseCode(commsSession.getResponseCode().getCode()); entity.setMessage(e.getMessage()); Response.ResponseBuilder builder = Response.status(Response.Status.BAD_REQUEST).entity(entity); return clusterContext(noCache(builder)).build(); } return responseCreator.unexpectedErrorResponse(portId, transactionId, e); } return clusterContext(noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager))).build(); } DataTransferResource(final NiFiProperties nifiProperties); @POST @Produces(MediaType.APPLICATION_JSON) @Path("{portType}/{portId}/transactions") @ApiOperation( value = "Create a transaction to the specified output port or input port", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/{component-type}/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response createPortTransaction(
            @ApiParam(
                    value = "The port type.",
                    required = true,
                    allowableValues = "input-ports, output-ports"
            )
            @PathParam("portType") String portType,
            @PathParam("portId") String portId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @POST @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.TEXT_PLAIN) @Path("input-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files to the input port", response = String.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response receiveFlowFiles(
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitOutputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "A checksum calculated at client side using CRC32 to check flow file content integrity. It must match with the value calculated at server side.",
                    required = true
            )
            @QueryParam(CHECK_SUM) @DefaultValue(StringUtils.EMPTY) String checksum,
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response commitInputPortTransaction(
            @ApiParam(
                    value = "The response code. Available values are BAD_CHECKSUM(19), CONFIRM_TRANSACTION(12) or CANCEL_TRANSACTION(15).",
                    required = true
            )
            @QueryParam(RESPONSE_CODE) Integer responseCode,
            @ApiParam(
                    value = "The input port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @ApiParam(
                    value = "The transaction id.",
                    required = true
            )
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context ServletContext context,
            InputStream inputStream); @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_OCTET_STREAM) @Path("output-ports/{portId}/transactions/{transactionId}/flow-files") @ApiOperation( value = "Transfer flow files from the output port", response = StreamingOutput.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @ApiResponses( value = { @ApiResponse(code = 200, message = "There is no flow file to return."), @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response transferFlowFiles(
            @ApiParam(
                    value = "The output port id.",
                    required = true
            )
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("input-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") } ) @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.") } ) Response extendInputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); @PUT @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_JSON) @Path("output-ports/{portId}/transactions/{transactionId}") @ApiOperation( value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") } ) @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."), @ApiResponse(code = 503, message = "NiFi instance is not ready for serving request, or temporarily overloaded. Retrying the same request later may be successful"), } ) Response extendOutputPortTransactionTTL(
            @PathParam("portId") String portId,
            @PathParam("transactionId") String transactionId,
            @Context HttpServletRequest req,
            @Context HttpServletResponse res,
            @Context ServletContext context,
            @Context UriInfo uriInfo,
            InputStream inputStream); Response extendPortTransactionTTL(
            String portType,
            String portId,
            String transactionId,
            HttpServletRequest req,
            HttpServletResponse res,
            ServletContext context,
            UriInfo uriInfo,
            InputStream inputStream); void setServiceFacade(NiFiServiceFacade serviceFacade); static final String CHECK_SUM; static final String RESPONSE_CODE; }