【发布时间】：2023-01-21 01:49:44
【问题描述】：
我有一个本地 Kinesalite 流，我尝试从我的本地 Kinesis 发送数据，它是在应用程序模式下构建的：
Properties sinkProperties = new Properties();
sinkProperties.put(AWSConfigConstants.AWS_REGION, region);
sinkProperties.put(AWSConfigConstants.AWS_SECRET_ACCESS_KEY, aws_secret_access_key);
sinkProperties.put(AWSConfigConstants.AWS_ACCESS_KEY_ID, aws_access_key);
sinkProperties.put(AWSConfigConstants.AWS_ENDPOINT, aws_endpoint);
sinkProperties.put(AWSConfigConstants.TRUST_ALL_CERTIFICATES, true);
DataStream<String> fromGen =
env.fromSequence(1, 10_000_000L)
.map(Object::toString)
.returns(String.class)
.map(data -> data.toString());
KinesisStreamsSink<String> kdsSink =
KinesisStreamsSink.<String>builder()
.setSerializationSchema(new SimpleStringSchema())
.setPartitionKeyGenerator(element -> String.valueOf(element.hashCode()))
.setStreamName(outputStreamName)
.setMaxBatchSize(20)
.setKinesisClientProperties(sinkProperties)
.build();
fromGen.sinkTo(kdsSink);
向该流发送新数据后出现以下错误：
15:48:53,933 WARN org.apache.flink.kinesis.shaded.io.netty.channel.DefaultChannelPipeline [] - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
org.apache.flink.kinesis.shaded.io.netty.handler.codec.DecoderException: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:477) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:986) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [flink-connector-kinesis-1.15.2.jar:1.15.2]
at java.lang.Thread.run(Thread.java:829) [?:?]
Caused by: javax.net.ssl.SSLHandshakeException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
at sun.security.ssl.Alert.createSSLException(Alert.java:131) ~[?:?]
at sun.security.ssl.TransportContext.fatal(TransportContext.java:353) ~[?:?]
at sun.security.ssl.TransportContext.fatal(TransportContext.java:296) ~[?:?]
at sun.security.ssl.TransportContext.fatal(TransportContext.java:291) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1357) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1232) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1175) ~[?:?]
at sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392) ~[?:?]
at sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061) ~[?:?]
at java.security.AccessController.doPrivileged(Native Method) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008) ~[?:?]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.runDelegatedTasks(SslHandler.java:1548) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1394) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1235) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1284) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
... 16 more
Caused by: sun.security.validator.ValidatorException: PKIX path building failed: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:439) ~[?:?]
at sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306) ~[?:?]
at sun.security.validator.Validator.validate(Validator.java:264) ~[?:?]
at sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313) ~[?:?]
at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:276) ~[?:?]
at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:141) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1335) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1232) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1175) ~[?:?]
at sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392) ~[?:?]
at sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061) ~[?:?]
at java.security.AccessController.doPrivileged(Native Method) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008) ~[?:?]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.runDelegatedTasks(SslHandler.java:1548) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1394) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1235) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1284) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
... 16 more
Caused by: sun.security.provider.certpath.SunCertPathBuilderException: unable to find valid certification path to requested target
at sun.security.provider.certpath.SunCertPathBuilder.build(SunCertPathBuilder.java:141) ~[?:?]
at sun.security.provider.certpath.SunCertPathBuilder.engineBuild(SunCertPathBuilder.java:126) ~[?:?]
at java.security.cert.CertPathBuilder.build(CertPathBuilder.java:297) ~[?:?]
at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:434) ~[?:?]
at sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:306) ~[?:?]
at sun.security.validator.Validator.validate(Validator.java:264) ~[?:?]
at sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:313) ~[?:?]
at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:276) ~[?:?]
at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:141) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.checkServerCerts(CertificateMessage.java:1335) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.onConsumeCertificate(CertificateMessage.java:1232) ~[?:?]
at sun.security.ssl.CertificateMessage$T13CertificateConsumer.consume(CertificateMessage.java:1175) ~[?:?]
at sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:392) ~[?:?]
at sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:443) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1074) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask$DelegatedAction.run(SSLEngineImpl.java:1061) ~[?:?]
at java.security.AccessController.doPrivileged(Native Method) ~[?:?]
at sun.security.ssl.SSLEngineImpl$DelegatedTask.run(SSLEngineImpl.java:1008) ~[?:?]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.runDelegatedTasks(SslHandler.java:1548) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1394) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1235) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1284) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
at org.apache.flink.kinesis.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446) ~[flink-connector-kinesis-1.15.2.jar:1.15.2]
... 16 more
这真的很奇怪，因为我能够从本地流中读取，只有当我尝试向它们添加新记录时才会出现此问题。
网上几乎没有任何资源如何解决这个问题，我找到了这个网站https://github.com/galgus/flink-connector-http，但解决方法对我不起作用（发生同样的错误）
【问题讨论】：
-
如果有人发现此线程：flink 应用程序版本与推荐的 aws flink 版本不同，并且旧版本的连接到 kinesis 的 sdk 存在漏洞。所以目前没有办法解决这个问题，我们选择在本地处理数据的不同方法
标签： apache-flink kinesis-stream