Support Questions

spserd · ‎02-22-2022

Hello all,

I am running into an issue with authentication when trying to connect my Flink (version 1.14.3) server with NiFi. NiFi 1.15.3 is running in secure mode in cluster with 3 nodes. When my Flink flow try to connect to NiFi, it throws:

Flink Log:

org.apache.flink.util.SerializedThrowable: Tried all cluster URLs but none of those was accessible. Last Exception was org.apache.nifi.remote.util.SiteToSiteRestApiClient$HttpGetFailedException: response code 401:Unauthorized with explanation: null
	at org.apache.nifi.remote.util.SiteToSiteRestApiClient.getController(SiteToSiteRestApiClient.java:372)
	at org.apache.nifi.remote.client.SiteInfoProvider.refreshRemoteInfo(SiteInfoProvider.java:69)
	at org.apache.nifi.remote.client.SiteInfoProvider.getPortIdentifier(SiteInfoProvider.java:220)
	at org.apache.nifi.remote.client.SiteInfoProvider.getOutputPortIdentifier(SiteInfoProvider.java:204)
	at org.apache.nifi.remote.client.socket.SocketClient.getPortIdentifier(SocketClient.java:79)
	at org.apache.nifi.remote.client.socket.SocketClient.createTransaction(SocketClient.java:121)
	at org.apache.flink.streaming.connectors.nifi.NiFiSource.run(NiFiSource.java:91)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:110)
	at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:67)
	at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:323)
Caused by: org.apache.flink.util.SerializedThrowable: response code 401:Unauthorized with explanation: null
	at org.apache.nifi.remote.util.SiteToSiteRestApiClient.execute(SiteToSiteRestApiClient.java:1203)
	at org.apache.nifi.remote.util.SiteToSiteRestApiClient.execute(SiteToSiteRestApiClient.java:1237)
	at org.apache.nifi.remote.util.SiteToSiteRestApiClient.fetchController(SiteToSiteRestApiClient.java:419)
	at org.apache.nifi.remote.util.SiteToSiteRestApiClient.getController(SiteToSiteRestApiClient.java:394)
	at org.apache.nifi.remote.util.SiteToSiteRestApiClient.getController(SiteToSiteRestApiClient.java:361)
	... 9 common frames omitted

In my NiFi logs:

2022-02-22 21:27:39,381 DEBUG [NiFi Web Server-1203] o.a.n.w.s.NiFiAuthenticationFilter Authenticating [null]
2022-02-22 21:27:39,381 DEBUG [NiFi Web Server-1203] o.a.n.w.s.x509.X509CertificateExtractor No client certificate found in request.
2022-02-22 21:27:39,381 DEBUG [NiFi Web Server-1203] o.a.n.w.s.NiFiAuthenticationFilter Authenticating [null]
2022-02-22 21:27:39,381 DEBUG [NiFi Web Server-1203] o.a.n.w.s.NiFiAuthenticationFilter Authenticating [null]
2022-02-22 21:27:39,381 INFO [NiFi Web Server-1203] o.a.n.w.s.NiFiAuthenticationFilter Authentication Started 16.13.8.6 [<anonymous>] GET https://nifi_node1:8443/nifi-api/site-to-site
2022-02-22 21:27:39,381 WARN [NiFi Web Server-1203] o.a.n.w.s.NiFiAuthenticationFilter Authentication Failed 16.13.8.6 GET https://nifi_node1:8443/nifi-api/site-to-site [Anonymous authentication has not been configured.]
2022-02-22 21:27:39,382 DEBUG [NiFi Web Server-1203] o.a.n.w.s.NiFiAuthenticationFilter Authentication Failed
org.apache.nifi.web.security.InvalidAuthenticationException: Anonymous authentication has not been configured.
	at org.apache.nifi.web.security.anonymous.NiFiAnonymousAuthenticationProvider.authenticate(NiFiAnonymousAuthenticationProvider.java:46)
	at org.springframework.security.authentication.ProviderManager.authenticate(ProviderManager.java:182)
	at org.apache.nifi.web.security.NiFiAuthenticationFilter.authenticate(NiFiAuthenticationFilter.java:73)
	at org.apache.nifi.web.security.NiFiAuthenticationFilter.doFilter(NiFiAuthenticationFilter.java:56)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:336)
	at org.apache.nifi.web.security.NiFiAuthenticationFilter.authenticate(NiFiAuthenticationFilter.java:94)
	at org.apache.nifi.web.security.NiFiAuthenticationFilter.doFilter(NiFiAuthenticationFilter.java:56)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:336)
	at org.springframework.security.oauth2.server.resource.web.BearerTokenAuthenticationFilter.doFilterInternal(BearerTokenAuthenticationFilter.java:121)
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:336)
	at org.apache.nifi.web.security.NiFiAuthenticationFilter.authenticate(NiFiAuthenticationFilter.java:94)
	at org.apache.nifi.web.security.NiFiAuthenticationFilter.doFilter(NiFiAuthenticationFilter.java:56)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:336)
	at org.springframework.security.web.csrf.CsrfFilter.doFilterInternal(CsrfFilter.java:117)
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:119)
	at org.springframework.security.web.FilterChainProxy$VirtualFilterChain.doFilter(FilterChainProxy.java:336)
	at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:211)
	at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:183)
	at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:358)
	at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:271)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.filter.TimerFilter.doFilter(TimerFilter.java:51)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.filter.ExceptionFilter.doFilter(ExceptionFilter.java:46)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:201)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.eclipse.jetty.servlets.DoSFilter.doFilterChain(DoSFilter.java:487)
	at org.eclipse.jetty.servlets.DoSFilter.doFilter(DoSFilter.java:336)
	at org.eclipse.jetty.servlets.DoSFilter.doFilter(DoSFilter.java:301)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.security.headers.StrictTransportSecurityFilter.doFilter(StrictTransportSecurityFilter.java:48)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.security.headers.XContentTypeOptionsFilter.doFilter(XContentTypeOptionsFilter.java:48)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.security.headers.XSSProtectionFilter.doFilter(XSSProtectionFilter.java:48)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.security.headers.ContentSecurityPolicyFilter.doFilter(ContentSecurityPolicyFilter.java:47)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.apache.nifi.web.security.headers.XFrameOptionsFilter.doFilter(XFrameOptionsFilter.java:48)
	at org.eclipse.jetty.servlet.FilterHolder.doFilter(FilterHolder.java:193)
	at org.eclipse.jetty.servlet.ServletHandler$Chain.doFilter(ServletHandler.java:1601)
	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:548)
	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:143)
	at org.eclipse.jetty.security.SecurityHandler.handle(SecurityHandler.java:600)
	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:235)
	at org.eclipse.jetty.server.session.SessionHandler.doHandle(SessionHandler.java:1624)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233)
	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1434)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188)
	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:501)
	at org.eclipse.jetty.server.session.SessionHandler.doScope(SessionHandler.java:1594)
	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186)
	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1349)
	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141)
	at org.eclipse.jetty.server.handler.HandlerCollection.handle(HandlerCollection.java:146)
	at org.eclipse.jetty.server.handler.gzip.GzipHandler.handle(GzipHandler.java:763)
	at org.eclipse.jetty.server.handler.ContextHandlerCollection.handle(ContextHandlerCollection.java:191)
	at org.eclipse.jetty.server.handler.HandlerList.handle(HandlerList.java:59)
	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127)
	at org.eclipse.jetty.server.Server.handle(Server.java:516)
	at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:400)
	at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:645)
	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:392)
	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:277)
	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311)
	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
	at org.eclipse.jetty.io.ssl.SslConnection$DecryptedEndPoint.onFillable(SslConnection.java:555)
	at org.eclipse.jetty.io.ssl.SslConnection.onFillable(SslConnection.java:410)
	at org.eclipse.jetty.io.ssl.SslConnection$2.succeeded(SslConnection.java:164)
	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105)
	at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:338)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:315)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:173)
	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:131)
	at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:409)
	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:883)
	at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:1034)
	at java.lang.Thread.run(Thread.java:748)

NiFi Properties (grepped for http and pasted the output):

nifi.remote.input.http.enabled=false
nifi.remote.input.http.transaction.ttl=30 sec
nifi.web.http.host=
nifi.web.http.port=
nifi.web.http.network.interface.default=
nifi.web.https.host=nifi_node1
nifi.web.https.port=8443
nifi.web.https.network.interface.default=
nifi.web.https.ciphersuites.include=
nifi.web.https.ciphersuites.exclude=
nifi.security.user.saml.signature.algorithm=http://www.w3.org/2001/04/xmldsig-more#rsa-sha256
nifi.security.user.saml.signature.digest.algorithm=http://www.w3.org/2001/04/xmlenc#sha256
nifi.security.user.saml.http.client.truststore.strategy=JDK
nifi.security.user.saml.http.client.connect.timeout=30 secs
nifi.security.user.saml.http.client.read.timeout=30 secs

I'm using same certificates in all NiFi nodes and Flink node. The certs are all in place and I confirmed there is one PrivateKeyEntry with ExtendedKeyUsages set to serverAuth and clientAuth. I searched for solution and I could not able to find one in the forum.

Note: The IP: 16.13.8.6 in the log is my Flink node

Please let me know what I'm missing.

Appreciate any help on this!

Thanks,
Prabu.

spserd · ‎03-03-2022

Hi @araujo,

I figured out the issue. It seems we have to provide the keystore type and truststore type along with the keystore and truststore file and passwords. Once I have them added to the source building code, I was able to connect to NiFi without any issues. Now my Flink flow is connected to NiFi and processing data.

Sharing the NiFiSource creation logic for reference:

org.apache.flink.streaming.api.functions.source.SourceFunction<org.apache.flink.streaming.connectors.nifi.NiFiDataPacket> nifiSource;
SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
        .urls(new HashSet(Arrays.asList(config.getString(clusterId + ".url").split(","))))
        .portName(config.getString(clusterId + ".port"))
        .requestBatchCount(config.getInt("batch-count"))
        .truststoreFilename(config.getString("truststore-file"))
        .truststorePass(config.getString("truststore-password"))
        .keystoreFilename(config.getString("keystore-file"))
        .keystorePass(config.getString("keystore-password"))
        .keystoreType(KeystoreType.JKS)
        .truststoreType(KeystoreType.JKS)
        .buildConfig();
nifiSource = new org.apache.flink.streaming.connectors.nifi.NiFiSource(clientConfig);

Appreciate your help and patience.

Thanks,

Prabu.

View solution in original post

spserd · ‎03-02-2022

Hi @araujo - Have you had a chance to look into this? Please let me know for your thoughts/suggestions.

Appreciate your help and patience.

Thanks,

Prabu.

PS: Please ignore the reply on 02-25-2022 at 10:41 AM which was incomplete. I don't see a delete option to delete it.

araujo · ‎03-02-2022

@spserd ,

Are the NiFi 1.11 and 1.15 clusters different clusters or the same cluster that you upgraded?

Is the application using the same keystore to connect to both clusters?

Would you be able to share the part of your application code where you construct your NiFiSource?

Cheers,

André

--
Was your question answered? Please take some time to click on "Accept as Solution" below this post.
If you find a reply useful, say thanks by clicking on the thumbs up button.

spserd · ‎03-03-2022

Hi @araujo,

I figured out the issue. It seems we have to provide the keystore type and truststore type along with the keystore and truststore file and passwords. Once I have them added to the source building code, I was able to connect to NiFi without any issues. Now my Flink flow is connected to NiFi and processing data.

Sharing the NiFiSource creation logic for reference:

org.apache.flink.streaming.api.functions.source.SourceFunction<org.apache.flink.streaming.connectors.nifi.NiFiDataPacket> nifiSource;
SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder()
        .urls(new HashSet(Arrays.asList(config.getString(clusterId + ".url").split(","))))
        .portName(config.getString(clusterId + ".port"))
        .requestBatchCount(config.getInt("batch-count"))
        .truststoreFilename(config.getString("truststore-file"))
        .truststorePass(config.getString("truststore-password"))
        .keystoreFilename(config.getString("keystore-file"))
        .keystorePass(config.getString("keystore-password"))
        .keystoreType(KeystoreType.JKS)
        .truststoreType(KeystoreType.JKS)
        .buildConfig();
nifiSource = new org.apache.flink.streaming.connectors.nifi.NiFiSource(clientConfig);

Appreciate your help and patience.

Thanks,

Prabu.

cjervis · ‎03-03-2022

@spserd

I'm happy to see you resolved your issue. Please mark the appropriate reply as the solution, as it will make it easier for others to find the answer in the future.

Cy Jervis, Manager, Community Program
Was your question answered? Make sure to mark the answer as the accepted solution.
If you find a reply useful, say thanks by clicking on the thumbs up button.

araujo · ‎03-03-2022

Ah! Thanks a lot for taking the time to post the solution, @spserd !

I suspect that there was one detail that wasn't mentioned: did you also upgrade Java from 8 to 11 between the two clusters?

In Java 8 the default keystore/truststore type used to be JKS and in Java 11 the default changed to PKCS12. If the store type isn't set explicitly, this error would certainly happen when upgrading Java.

Cheers,

André

--
Was your question answered? Please take some time to click on "Accept as Solution" below this post.
If you find a reply useful, say thanks by clicking on the thumbs up button.

spserd · ‎03-04-2022

Hello @araujo - Yes. That is the issue. From Java 11, we may need to explicitly set the keystore type.

Thanks,

Prabu

Cloudera Community

Support Questions

Authentication Issue: X509CertificateExtractor No client certificate found in request.