http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/DataTransferResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/DataTransferResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/DataTransferResource.java index d5e7d36..38a738c 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/DataTransferResource.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/DataTransferResource.java @@ -92,6 +92,7 @@ import static org.apache.nifi.remote.protocol.http.HttpHeaders.HANDSHAKE_PROPERT import static org.apache.nifi.remote.protocol.http.HttpHeaders.HANDSHAKE_PROPERTY_BATCH_SIZE; import static org.apache.nifi.remote.protocol.http.HttpHeaders.HANDSHAKE_PROPERTY_REQUEST_EXPIRATION; import static org.apache.nifi.remote.protocol.http.HttpHeaders.HANDSHAKE_PROPERTY_USE_COMPRESSION; +import org.apache.nifi.util.NiFiProperties; /** * RESTful endpoint for managing a SiteToSite connection. @@ -108,14 +109,19 @@ public class DataTransferResource extends ApplicationResource { public static final String CHECK_SUM = "checksum"; public static final String RESPONSE_CODE = "responseCode"; - private static final String PORT_TYPE_INPUT = "input-ports"; private static final String PORT_TYPE_OUTPUT = "output-ports"; private Authorizer authorizer; private final ResponseCreator responseCreator = new ResponseCreator(); private final VersionNegotiator transportProtocolVersionNegotiator = new TransportProtocolVersionNegotiator(1); - private final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + private final HttpRemoteSiteListener transactionManager; + private final NiFiProperties nifiProperties; + + public DataTransferResource(final NiFiProperties nifiProperties) { + this.nifiProperties = nifiProperties; + transactionManager = HttpRemoteSiteListener.getInstance(nifiProperties); + } /** * Authorizes access to data transfers. @@ -161,18 +167,17 @@ public class DataTransferResource extends ApplicationResource { 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 = "") + @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"), - } + @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 createPortTransaction( @ApiParam( @@ -187,7 +192,6 @@ public class DataTransferResource extends ApplicationResource { @Context UriInfo uriInfo, InputStream inputStream) { - if (!PORT_TYPE_INPUT.equals(portType) && !PORT_TYPE_OUTPUT.equals(portType)) { return responseCreator.wrongPortTypeResponse(portType, portId); } @@ -235,18 +239,17 @@ public class DataTransferResource extends ApplicationResource { value = "Transfer flow files to the input port", response = String.class, authorizations = { - @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") + @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"), - } + @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 receiveFlowFiles( @ApiParam( @@ -299,26 +302,26 @@ public class DataTransferResource extends ApplicationResource { } private HttpFlowFileServerProtocol initiateServerProtocol(final HttpServletRequest req, final Peer peer, - final Integer transportProtocolVersion) throws IOException { + final Integer transportProtocolVersion) throws IOException { // Switch transaction protocol version based on transport protocol version. TransportProtocolVersionNegotiator negotiatedTransportProtocolVersion = new TransportProtocolVersionNegotiator(transportProtocolVersion); VersionNegotiator versionNegotiator = new StandardVersionNegotiator(negotiatedTransportProtocolVersion.getTransactionProtocolVersion()); final String dataTransferUrl = req.getRequestURL().toString(); - ((HttpCommunicationsSession)peer.getCommunicationsSession()).setDataTransferUrl(dataTransferUrl); + ((HttpCommunicationsSession) peer.getCommunicationsSession()).setDataTransferUrl(dataTransferUrl); HttpFlowFileServerProtocol serverProtocol = getHttpFlowFileServerProtocol(versionNegotiator); - HttpRemoteSiteListener.getInstance().setupServerProtocol(serverProtocol); + HttpRemoteSiteListener.getInstance(nifiProperties).setupServerProtocol(serverProtocol); serverProtocol.handshake(peer); return serverProtocol; } HttpFlowFileServerProtocol getHttpFlowFileServerProtocol(final VersionNegotiator versionNegotiator) { - return new StandardHttpFlowFileServerProtocol(versionNegotiator); + return new StandardHttpFlowFileServerProtocol(versionNegotiator, nifiProperties); } private Peer constructPeer(final HttpServletRequest req, final InputStream inputStream, - final OutputStream outputStream, final String portId, final String transactionId) { + final OutputStream outputStream, final String portId, final String transactionId) { final String clientHostName = req.getRemoteHost(); final int clientPort = req.getRemotePort(); @@ -376,18 +379,17 @@ public class DataTransferResource extends ApplicationResource { value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { - @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") + @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"), - } + @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( @@ -474,7 +476,6 @@ public class DataTransferResource extends ApplicationResource { return clusterContext(noCache(setCommonHeaders(Response.ok(entity), transportProtocolVersion, transactionManager))).build(); } - @DELETE @Consumes(MediaType.APPLICATION_OCTET_STREAM) @Produces(MediaType.APPLICATION_JSON) @@ -483,18 +484,17 @@ public class DataTransferResource extends ApplicationResource { value = "Commit or cancel the specified transaction", response = TransactionResultEntity.class, authorizations = { - @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") + @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"), - } + @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 commitInputPortTransaction( @ApiParam( @@ -592,7 +592,6 @@ public class DataTransferResource extends ApplicationResource { return Response.ok(entity).build(); } - @GET @Consumes(MediaType.WILDCARD) @Produces(MediaType.APPLICATION_OCTET_STREAM) @@ -601,19 +600,18 @@ public class DataTransferResource extends ApplicationResource { value = "Transfer flow files from the output port", response = StreamingOutput.class, authorizations = { - @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") + @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"), - } + @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"),} ) public Response transferFlowFiles( @ApiParam( @@ -685,16 +683,16 @@ public class DataTransferResource extends ApplicationResource { value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { - @Authorization(value = "Write - /data-transfer/input-ports/{uuid}", type = "") + @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 = 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 extendInputPortTransactionTTL( @@ -720,18 +718,17 @@ public class DataTransferResource extends ApplicationResource { value = "Extend transaction TTL", response = TransactionResultEntity.class, authorizations = { - @Authorization(value = "Write - /data-transfer/output-ports/{uuid}", type = "") + @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"), - } + @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 extendOutputPortTransactionTTL( @PathParam("portId") String portId, @@ -794,6 +791,7 @@ public class DataTransferResource extends ApplicationResource { } private class ValidateRequestResult { + private Integer transportProtocolVersion; private Response errResponse; } @@ -824,9 +822,7 @@ public class DataTransferResource extends ApplicationResource { return result; } - // setters - public void setAuthorizer(Authorizer authorizer) { this.authorizer = authorizer; }
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/SiteToSiteResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/SiteToSiteResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/SiteToSiteResource.java index 66b1151..88bdeb6 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/SiteToSiteResource.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/SiteToSiteResource.java @@ -16,23 +16,9 @@ */ package org.apache.nifi.web.api; -import static org.apache.commons.lang3.StringUtils.isEmpty; import java.net.InetAddress; import java.net.UnknownHostException; -import java.util.ArrayList; -import java.util.List; -import java.util.Set; - -import javax.servlet.http.HttpServletRequest; -import javax.ws.rs.Consumes; -import javax.ws.rs.GET; -import javax.ws.rs.HttpMethod; -import javax.ws.rs.Path; -import javax.ws.rs.Produces; -import javax.ws.rs.core.Context; -import javax.ws.rs.core.MediaType; -import javax.ws.rs.core.Response; import com.wordnik.swagger.annotations.Api; import com.wordnik.swagger.annotations.ApiOperation; @@ -79,6 +65,7 @@ import java.util.List; import java.util.Set; import static org.apache.commons.lang3.StringUtils.isEmpty; +import org.apache.nifi.util.NiFiProperties; /** * RESTful endpoint for managing a SiteToSite connection. @@ -98,7 +85,11 @@ public class SiteToSiteResource extends ApplicationResource { private final ResponseCreator responseCreator = new ResponseCreator(); private final VersionNegotiator transportProtocolVersionNegotiator = new TransportProtocolVersionNegotiator(1); - private final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + private final HttpRemoteSiteListener transactionManager; + + public SiteToSiteResource(final NiFiProperties nifiProperties) { + transactionManager = HttpRemoteSiteListener.getInstance(nifiProperties); + } /** * Authorizes access to Site To Site details. @@ -135,15 +126,15 @@ public class SiteToSiteResource extends ApplicationResource { value = "Returns the details about this NiFi necessary to communicate via site to site", response = ControllerEntity.class, authorizations = { - @Authorization(value = "Read - /site-to-site", type = "") + @Authorization(value = "Read - /site-to-site", 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 = 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 = 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 = 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 getSiteToSiteDetails(@Context HttpServletRequest req) { @@ -173,7 +164,6 @@ public class SiteToSiteResource extends ApplicationResource { return clusterContext(noCache(Response.ok(entity))).build(); } - /** * Returns the available Peers and its status of this NiFi. * @@ -188,15 +178,15 @@ public class SiteToSiteResource extends ApplicationResource { value = "Returns the available Peers and its status of this NiFi", response = PeersEntity.class, authorizations = { - @Authorization(value = "Read - /site-to-site", type = "") + @Authorization(value = "Read - /site-to-site", 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 = 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 = 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 = 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 getPeers(@Context HttpServletRequest req) { @@ -261,7 +251,6 @@ public class SiteToSiteResource extends ApplicationResource { } // setters - public void setServiceFacade(final NiFiServiceFacade serviceFacade) { this.serviceFacade = serviceFacade; } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/AccessControlHelper.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/AccessControlHelper.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/AccessControlHelper.java index f740af7..b6f419f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/AccessControlHelper.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/AccessControlHelper.java @@ -26,6 +26,8 @@ import org.apache.nifi.nar.NarClassLoaders; import org.apache.nifi.util.NiFiProperties; import java.io.File; +import java.util.HashMap; +import java.util.Map; import static org.junit.Assert.assertEquals; @@ -68,12 +70,12 @@ public class AccessControlHelper { System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, nifiPropertiesFile.getAbsolutePath()); // update the flow.xml property - NiFiProperties props = NiFiProperties.getInstance(); - props.setProperty(NiFiProperties.FLOW_CONFIGURATION_FILE, flowXmlPath); - + final Map<String, String> addProps = new HashMap<>(); + addProps.put(NiFiProperties.FLOW_CONFIGURATION_FILE, flowXmlPath); if (overrideAuthorizer != null) { - props.setProperty(NiFiProperties.SECURITY_USER_AUTHORIZER, overrideAuthorizer); + addProps.put(NiFiProperties.SECURITY_USER_AUTHORIZER, overrideAuthorizer); } + NiFiProperties props = NiFiProperties.createBasicNiFiProperties(null, addProps); // load extensions NarClassLoaders.getInstance().init(props.getFrameworkWorkingDirectory(), props.getExtensionsWorkingDirectory()); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/ITAccessTokenEndpoint.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/ITAccessTokenEndpoint.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/ITAccessTokenEndpoint.java index 3336669..dc07fdf 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/ITAccessTokenEndpoint.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/accesscontrol/ITAccessTokenEndpoint.java @@ -64,8 +64,9 @@ public class ITAccessTokenEndpoint { System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, nifiPropertiesFile.getAbsolutePath()); // update the flow.xml property - NiFiProperties props = NiFiProperties.getInstance(); - props.setProperty("nifi.flow.configuration.file", FLOW_XML_PATH); + final Map<String, String> addProps = new HashMap<>(); + addProps.put("nifi.flow.configuration.file", FLOW_XML_PATH); + NiFiProperties props = NiFiProperties.createBasicNiFiProperties(null, addProps); // delete the database directory to avoid issues with re-registration in testRequestAccessUsingToken FileUtils.deleteDirectory(props.getDatabaseRepositoryPath().toFile()); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/util/NiFiTestServer.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/util/NiFiTestServer.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/util/NiFiTestServer.java index 4c1a417..55c3168 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/util/NiFiTestServer.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/integration/util/NiFiTestServer.java @@ -18,6 +18,8 @@ package org.apache.nifi.integration.util; import com.sun.jersey.api.client.Client; import java.io.File; +import java.net.URISyntaxException; +import java.net.URL; import java.util.Collections; import javax.servlet.ServletContext; import org.apache.nifi.util.NiFiProperties; @@ -50,8 +52,15 @@ public class NiFiTestServer { private WebAppContext webappContext; public NiFiTestServer(String webappRoot, String contextPath) { - // load the configuration - properties = NiFiProperties.getInstance(); + final URL resource = NiFiTestServer.class.getResource("/site-to-site/nifi.properties"); + try { + final String propertiesFile = resource.toURI().getPath(); + System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, propertiesFile); + // load the configuration + properties = NiFiProperties.createBasicNiFiProperties(null, null); + } catch (final URISyntaxException ue) { + throw new RuntimeException(); + } createWebAppContext(webappRoot, contextPath); createServer(); @@ -171,14 +180,16 @@ public class NiFiTestServer { } /** - * Convenience method to provide access to Spring beans accessible from the web application context. + * Convenience method to provide access to Spring beans accessible from the + * web application context. * * @param <T> target cast * @param beanName name of the spring bean * @param clazz class of the spring bean * @return Spring bean with given name and class type * - * @throws ClassCastException if the bean found cannot be cast to the given class type + * @throws ClassCastException if the bean found cannot be cast to the given + * class type */ public <T> T getSpringBean(String beanName, Class<T> clazz) { ServletContext servletContext = webappContext.getServletHandler().getServletContext(); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestDataTransferResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestDataTransferResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestDataTransferResource.java index 422dbc3..eb87830 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestDataTransferResource.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestDataTransferResource.java @@ -178,7 +178,7 @@ public class TestDataTransferResource { final UriInfo uriInfo = mockUriInfo(locationUriStr); final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.extendPortTransactionTTL("input-ports", "port-id", transactionId, req, res, context, uriInfo, inputStream); @@ -209,7 +209,7 @@ public class TestDataTransferResource { final ServletContext context = null; final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.receiveFlowFiles("port-id", transactionId, req, context, inputStream); @@ -236,7 +236,7 @@ public class TestDataTransferResource { final ServletContext context = null; final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.receiveFlowFiles("port-id", transactionId, req, context, inputStream); @@ -255,7 +255,7 @@ public class TestDataTransferResource { final ServletContext context = null; final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.commitInputPortTransaction(ResponseCode.CONFIRM_TRANSACTION.getCode(), "port-id", transactionId, req, context, inputStream); @@ -278,7 +278,7 @@ public class TestDataTransferResource { final HttpServletResponse res = null; final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.transferFlowFiles("port-id", transactionId, req, res, context, inputStream); @@ -300,7 +300,7 @@ public class TestDataTransferResource { final ServletContext context = null; final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.commitOutputPortTransaction(ResponseCode.CONFIRM_TRANSACTION.getCode(), @@ -326,7 +326,7 @@ public class TestDataTransferResource { final ServletContext context = null; final InputStream inputStream = null; - final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(); + final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null)); final String transactionId = transactionManager.createTransaction(); final Response response = resource.commitOutputPortTransaction(ResponseCode.CONFIRM_TRANSACTION.getCode(), @@ -342,7 +342,7 @@ public class TestDataTransferResource { private DataTransferResource getDataTransferResource() { final HttpFlowFileServerProtocol serverProtocol = mock(HttpFlowFileServerProtocol.class); - final DataTransferResource resource = new DataTransferResource() { + final DataTransferResource resource = new DataTransferResource(NiFiProperties.createBasicNiFiProperties(null, null)) { @Override protected void authorizeDataTransfer(ResourceType resourceType, String identifier) { } @@ -352,7 +352,7 @@ public class TestDataTransferResource { return serverProtocol; } }; - resource.setProperties(NiFiProperties.getInstance()); + resource.setProperties(NiFiProperties.createBasicNiFiProperties(null, null)); return resource; } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestSiteToSiteResource.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestSiteToSiteResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestSiteToSiteResource.java index 8b65c3a..60a7ba9 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestSiteToSiteResource.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/test/java/org/apache/nifi/web/api/TestSiteToSiteResource.java @@ -151,12 +151,12 @@ public class TestSiteToSiteResource { } private SiteToSiteResource getSiteToSiteResource(final NiFiServiceFacade serviceFacade) { - final SiteToSiteResource resource = new SiteToSiteResource() { + final SiteToSiteResource resource = new SiteToSiteResource(NiFiProperties.createBasicNiFiProperties(null, null)) { @Override protected void authorizeSiteToSite() { } }; - resource.setProperties(NiFiProperties.getInstance()); + resource.setProperties(NiFiProperties.createBasicNiFiProperties(null, null)); resource.setServiceFacade(serviceFacade); return resource; } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/groovy/org/apache/nifi/web/security/x509/ocsp/OcspCertificateValidatorGroovyTest.groovy ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/groovy/org/apache/nifi/web/security/x509/ocsp/OcspCertificateValidatorGroovyTest.groovy b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/groovy/org/apache/nifi/web/security/x509/ocsp/OcspCertificateValidatorGroovyTest.groovy index 575800c..e3ff000 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/groovy/org/apache/nifi/web/security/x509/ocsp/OcspCertificateValidatorGroovyTest.groovy +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/groovy/org/apache/nifi/web/security/x509/ocsp/OcspCertificateValidatorGroovyTest.groovy @@ -65,7 +65,17 @@ public class OcspCertificateValidatorGroovyTest { @Before public void setUp() throws Exception { - mockProperties = [getProperty: { String propertyName -> return "value_for_${propertyName}" }] as NiFiProperties + mockProperties = new NiFiProperties() { + @Override + String getProperty(String key) { + return 'value_for_' + key + } + + @Override + Set<String> getPropertyKeys() { + return ["A", "set"] + } + } } @After http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/java/org/apache/nifi/web/security/NiFiAuthenticationProviderTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/java/org/apache/nifi/web/security/NiFiAuthenticationProviderTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/java/org/apache/nifi/web/security/NiFiAuthenticationProviderTest.java index 0e25747..eb89c22 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/java/org/apache/nifi/web/security/NiFiAuthenticationProviderTest.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-security/src/test/java/org/apache/nifi/web/security/NiFiAuthenticationProviderTest.java @@ -170,7 +170,7 @@ public class NiFiAuthenticationProviderTest { private NiFiProperties getNiFiProperties(final Properties properties) { final NiFiProperties nifiProperties = Mockito.mock(NiFiProperties.class); - when(nifiProperties.stringPropertyNames()).thenReturn(properties.stringPropertyNames()); + when(nifiProperties.getPropertyKeys()).thenReturn(properties.stringPropertyNames()); when(nifiProperties.getProperty(anyString())).then(new Answer<String>() { @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java index 93e0703..09450c2 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/AbstractHadoopProcessor.java @@ -41,7 +41,6 @@ import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; -import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.StringUtils; import javax.net.SocketFactory; @@ -128,6 +127,9 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor { private long lastKerberosReloginTime; protected KerberosProperties kerberosProperties; protected List<PropertyDescriptor> properties; + private volatile String kerberosServicePrincipal = null; + private volatile File kerberosConfigFile = null; + private volatile File kerberosServiceKeytab = null; // variables shared by all threads of this processor // Hadoop Configuration, Filesystem, and UserGroupInformation (optional) @@ -147,10 +149,13 @@ public abstract class AbstractHadoopProcessor extends AbstractProcessor { props.add(kerberosProperties.getKerberosKeytab()); props.add(KERBEROS_RELOGIN_PERIOD); properties = Collections.unmodifiableList(props); + kerberosServicePrincipal = context.getKerberosServicePrincipal(); + kerberosConfigFile = context.getKerberosConfigurationFile(); + kerberosServiceKeytab = context.getKerberosServiceKeytab(); } protected KerberosProperties getKerberosProperties() { - return KerberosProperties.create(NiFiProperties.getInstance()); + return new KerberosProperties(kerberosConfigFile); } @Override http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/AbstractHadoopTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/AbstractHadoopTest.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/AbstractHadoopTest.java index 9e2193d..80d3867 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/AbstractHadoopTest.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/AbstractHadoopTest.java @@ -67,7 +67,7 @@ public class AbstractHadoopTest { // mock properties and return a temporary file for the kerberos configuration mockedProperties = mock(NiFiProperties.class); when(mockedProperties.getKerberosConfigurationFile()).thenReturn(temporaryFile); - kerberosProperties = KerberosProperties.create(mockedProperties); + kerberosProperties = new KerberosProperties(temporaryFile); } @After @@ -139,8 +139,7 @@ public class AbstractHadoopTest { @Test public void testKerberosOptionsWithBadKerberosConfigFile() throws Exception { // invalid since the kerberos configuration was changed to a non-existent file - when(mockedProperties.getKerberosConfigurationFile()).thenReturn(new File("BAD_KERBEROS_PATH")); - kerberosProperties = KerberosProperties.create(mockedProperties); + kerberosProperties = new KerberosProperties(new File("BAD_KERBEROS_PATH")); SimpleHadoopProcessor processor = new SimpleHadoopProcessor(kerberosProperties); TestRunner runner = TestRunners.newTestRunner(processor); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java index 582346a..13c5436 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/GetHDFSTest.java @@ -50,7 +50,7 @@ public class GetHDFSTest { public void setup() { mockNiFiProperties = mock(NiFiProperties.class); when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null); - kerberosProperties = KerberosProperties.create(mockNiFiProperties); + kerberosProperties = new KerberosProperties(null); } @Test http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java index c8f8fb1..72c7da9 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/PutHDFSTest.java @@ -73,7 +73,7 @@ public class PutHDFSTest { public void setup() { mockNiFiProperties = mock(NiFiProperties.class); when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null); - kerberosProperties = KerberosProperties.create(mockNiFiProperties); + kerberosProperties = new KerberosProperties(null); } @Test http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestCreateHadoopSequenceFile.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestCreateHadoopSequenceFile.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestCreateHadoopSequenceFile.java index 8164fc0..af5fb3b 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestCreateHadoopSequenceFile.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestCreateHadoopSequenceFile.java @@ -70,7 +70,7 @@ public class TestCreateHadoopSequenceFile { public void setUp() { mockNiFiProperties = mock(NiFiProperties.class); when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null); - kerberosProperties = KerberosProperties.create(mockNiFiProperties); + kerberosProperties = new KerberosProperties(null); CreateHadoopSequenceFile proc = new TestableCreateHadoopSequenceFile(kerberosProperties); controller = TestRunners.newTestRunner(proc); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java index e49975b..bbfb095 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestFetchHDFS.java @@ -41,7 +41,7 @@ public class TestFetchHDFS { public void setup() { mockNiFiProperties = mock(NiFiProperties.class); when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null); - kerberosProperties = KerberosProperties.create(mockNiFiProperties); + kerberosProperties = new KerberosProperties(null); proc = new TestableFetchHDFS(kerberosProperties); runner = TestRunners.newTestRunner(proc); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java index d4204ea..a765572 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java @@ -68,7 +68,7 @@ public class TestListHDFS { public void setup() throws InitializationException { mockNiFiProperties = mock(NiFiProperties.class); when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null); - kerberosProperties = KerberosProperties.create(mockNiFiProperties); + kerberosProperties = new KerberosProperties(null); proc = new ListHDFSWithMockedFileSystem(kerberosProperties); runner = TestRunners.newTestRunner(proc); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/inotify/TestGetHDFSEvents.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/inotify/TestGetHDFSEvents.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/inotify/TestGetHDFSEvents.java index f8c1cce..939cbfc 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/inotify/TestGetHDFSEvents.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/inotify/TestGetHDFSEvents.java @@ -57,7 +57,7 @@ public class TestGetHDFSEvents { public void setup() { mockNiFiProperties = mock(NiFiProperties.class); when(mockNiFiProperties.getKerberosConfigurationFile()).thenReturn(null); - kerberosProperties = KerberosProperties.create(mockNiFiProperties); + kerberosProperties = new KerberosProperties(null); inotifyEventInputStream = mock(DFSInotifyEventInputStream.class); hdfsAdmin = mock(HdfsAdmin.class); } http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java index c2ec43b..0f79002 100644 --- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java +++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/dbcp/hive/HiveConnectionPool.java @@ -16,6 +16,7 @@ */ package org.apache.nifi.dbcp.hive; +import java.io.File; import org.apache.commons.dbcp.BasicDataSource; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.security.UserGroupInformation; @@ -35,7 +36,6 @@ import org.apache.nifi.logging.ComponentLog; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; import org.apache.nifi.reporting.InitializationException; -import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.hive.AuthenticationFailedException; import org.apache.nifi.util.hive.HiveConfigurator; import org.apache.nifi.util.hive.HiveUtils; @@ -47,14 +47,15 @@ import java.sql.Connection; import java.sql.SQLException; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import org.apache.nifi.controller.ControllerServiceInitializationContext; /** - * Implementation for Database Connection Pooling Service used for Apache Hive connections. Apache DBCP is used for connection pooling functionality. + * Implementation for Database Connection Pooling Service used for Apache Hive + * connections. Apache DBCP is used for connection pooling functionality. */ @Tags({"hive", "dbcp", "jdbc", "database", "connection", "pooling", "store"}) @CapabilityDescription("Provides Database Connection Pooling Service for Apache Hive. Connections can be asked from pool and returned after usage.") @@ -122,9 +123,8 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv private static final long TICKET_RENEWAL_PERIOD = 60000; private final static List<PropertyDescriptor> properties; - private static KerberosProperties kerberosProperties; - private String connectionUrl = "unknown"; + private String connectionUrl = "unknown"; // Holder of cached Configuration information so validation does not reload the same config over and over private final AtomicReference<ValidationResources> validationResourceHolder = new AtomicReference<>(); @@ -133,21 +133,31 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv private volatile HiveConfigurator hiveConfigurator = new HiveConfigurator(); private volatile UserGroupInformation ugi; + private volatile String kerberosServicePrincipal = null; + private volatile File kerberosConfigFile = null; + private volatile File kerberosServiceKeytab = null; + private volatile KerberosProperties kerberosProperties; static { - kerberosProperties = KerberosProperties.create(NiFiProperties.getInstance()); List<PropertyDescriptor> props = new ArrayList<>(); props.add(DATABASE_URL); props.add(HIVE_CONFIGURATION_RESOURCES); - props.add(kerberosProperties.getKerberosPrincipal()); - props.add(kerberosProperties.getKerberosKeytab()); props.add(DB_USER); props.add(DB_PASSWORD); props.add(MAX_WAIT_TIME); props.add(MAX_TOTAL_CONNECTIONS); - properties = Collections.unmodifiableList(props); + properties = props; } + @Override + protected void init(final ControllerServiceInitializationContext context) { + kerberosServicePrincipal = context.getKerberosServicePrincipal(); + kerberosConfigFile = context.getKerberosConfigurationFile(); + kerberosServiceKeytab = context.getKerberosServiceKeytab(); + kerberosProperties = new KerberosProperties(kerberosConfigFile); + properties.add(kerberosProperties.getKerberosPrincipal()); + properties.add(kerberosProperties.getKerberosKeytab()); + } @Override protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { @@ -213,7 +223,6 @@ public class HiveConnectionPool extends AbstractControllerService implements Hiv } getLogger().info("Successfully logged in as principal {} with keytab {}", new Object[]{principal, keyTab}); - } final String user = context.getProperty(DB_USER).getValue(); final String passw = context.getProperty(DB_PASSWORD).getValue(); http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java index c14d0aa..fdff58b 100644 --- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java +++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/main/java/org/apache/nifi/processors/hive/PutHiveStreaming.java @@ -52,8 +52,6 @@ import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; -import org.apache.nifi.util.NiFiProperties; -import org.apache.nifi.util.StringUtils; import org.apache.nifi.util.hive.AuthenticationFailedException; import org.apache.nifi.util.hive.HiveConfigurator; import org.apache.nifi.util.hive.HiveOptions; @@ -367,7 +365,7 @@ public class PutHiveStreaming extends AbstractProcessor { final List<String> partitionColumnList; final String partitionColumns = context.getProperty(PARTITION_COLUMNS).getValue(); - if (StringUtils.isEmpty(partitionColumns)) { + if (partitionColumns == null || partitionColumns.isEmpty()) { partitionColumnList = Collections.emptyList(); } else { String[] partitionCols = partitionColumns.split(","); @@ -832,7 +830,7 @@ public class PutHiveStreaming extends AbstractProcessor { } protected KerberosProperties getKerberosProperties() { - return KerberosProperties.create(NiFiProperties.getInstance()); + return kerberosProperties; } protected class HiveStreamingRecord { http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java index f95a377..f32c301 100644 --- a/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java +++ b/nifi-nar-bundles/nifi-hive-bundle/nifi-hive-processors/src/test/java/org/apache/nifi/processors/hive/TestPutHiveStreaming.java @@ -34,7 +34,6 @@ import org.apache.hive.hcatalog.streaming.TransactionBatch; import org.apache.nifi.hadoop.KerberosProperties; import org.apache.nifi.stream.io.ByteArrayOutputStream; import org.apache.nifi.util.MockFlowFile; -import org.apache.nifi.util.NiFiProperties; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; import org.apache.nifi.util.hive.HiveOptions; @@ -60,7 +59,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; /** * Unit tests for PutHiveStreaming processor. @@ -81,13 +79,9 @@ public class TestPutHiveStreaming { System.setProperty("java.security.krb5.realm", "nifi.com"); System.setProperty("java.security.krb5.kdc", "nifi.kdc"); - NiFiProperties niFiPropertiesWithKerberos = mock(NiFiProperties.class); - when(niFiPropertiesWithKerberos.getKerberosConfigurationFile()).thenReturn(new File("src/test/resources/krb5.conf")); - kerberosPropsWithFile = KerberosProperties.create(niFiPropertiesWithKerberos); + kerberosPropsWithFile = new KerberosProperties(new File("src/test/resources/krb5.conf")); - NiFiProperties niFiPropertiesWithoutKerberos = mock(NiFiProperties.class); - when(niFiPropertiesWithKerberos.getKerberosConfigurationFile()).thenReturn(null); - kerberosPropsWithoutFile = KerberosProperties.create(niFiPropertiesWithoutKerberos); + kerberosPropsWithoutFile = new KerberosProperties(null); processor = new MockPutHiveStreaming(); processor.setKerberosProperties(kerberosPropsWithFile); @@ -520,6 +514,7 @@ public class TestPutHiveStreaming { } private class MockPutHiveStreaming extends PutHiveStreaming { + private KerberosProperties kerberosProperties; private boolean generateConnectFailure = false; private boolean generateInterruptedExceptionOnCreateWriter = false; @@ -603,7 +598,7 @@ public class TestPutHiveStreaming { private HiveEndPoint endPoint; public MockHiveWriter(HiveEndPoint endPoint, int txnsPerBatch, boolean autoCreatePartitions, - long callTimeout, ExecutorService callTimeoutPool, UserGroupInformation ugi) + long callTimeout, ExecutorService callTimeoutPool, UserGroupInformation ugi) throws InterruptedException, ConnectFailure { super(endPoint, txnsPerBatch, autoCreatePartitions, callTimeout, callTimeoutPool, ugi); this.endPoint = endPoint; @@ -714,5 +709,4 @@ public class TestPutHiveStreaming { } } - -} \ No newline at end of file +}
