Соединитель приемника DataStax (kafka-connect-dse) выбрасывает javax. net .ssl.SSLException: SSLEngine уже закрыт - PullRequest
Новая
       67

Соединитель приемника DataStax (kafka-connect-dse) выбрасывает javax. net .ssl.SSLException: SSLEngine уже закрыт

0 голосов
/ 27 апреля 2020

Мы хотим, чтобы соединитель приемника DSE был подключен к (DSE-5.1.2) версии выпуска cassandra (3.11.3.5112). У нас есть отдельные linux коробки для kafka, kafka-connect и cassandra. Мы добавили kafka-connect-dse-1.3.1.jar в качестве плагина для kafka-connect, и ниже конфигурации, которую мы регистрируем для соединителя: 

{
"name": "dse-sink-connector-for-orders",
"config":   {"connector.class":"com.datastax.kafkaconnector.DseSinkConnector", 
"topics":"action_log", "tasks.max":"1", 
"ssl.provider":"JDK",  
"ssl.truststore.path":"<<Path_To_trust_store>>", 
"ssl.truststore.password":"<<password>>", 
"auth.provider":"DSE", "auth.username":"<<username>>", 
"auth.password":"<<password>>",  
"contactPoints":"172.21.96.64", 
"loadBalancing.localDc":"DC1", "port":"9042", 
"topic.action_log.testc_netspend.dse_orders.consistencyLevel": "LOCAL_ONE", 
"topic.action_log.testc_netspend.dse_orders.mapping":
"id=value.ACTION_ID,created=value.USER_ID, price=value.ACTION_ID, product=value.USER_ID, qty=value.ACTION_ID" 
} 
}

Когда мы нажимаем https://172.21.96.64: 8083 / connectors / dse-sink-connector-for-orders / status , это показывает, что соединитель работает, но задача не выполнена. Мы нашли следующую ошибку в журналах: 

[dse-sink-connector-for-orders|task-0] WorkerSinkTask{id=dse-sink-connector-for-orders-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:179)
com.datastax.oss.driver.api.core.AllNodesFailedException: Could not reach any contact point, make sure you've provided valid addresses (showing first 3 nodes, use getAllErrors() for more): Node(endPoint=test-db-cass0a02dc1/172.21.96.64:9042, hostId=null, hashCode=5fa99a2e): [com.datastax.oss.driver.api.core.connection.ConnectionInitException: [s1|control|id: 0xc4c37855, L:/172.21.117.41:52221 - R:test-db-cass0a02dc1/172.21.96.64:9042] Protocol initialization request, step 1 (OPTIONS): failed to send request (javax.net.ssl.SSLException: SSLEngine closed already)]
    at com.datastax.oss.driver.api.core.AllNodesFailedException.copy(AllNodesFailedException.java:141)
    at com.datastax.oss.driver.internal.core.util.concurrent.CompletableFutures.getUninterruptibly(CompletableFutures.java:149)
    at com.datastax.oss.driver.api.core.session.SessionBuilder.build(SessionBuilder.java:612)
    at com.datastax.kafkaconnector.state.LifeCycleManager.buildCqlSession(LifeCycleManager.java:521)
    at com.datastax.kafkaconnector.state.LifeCycleManager.lambda$startTask$0(LifeCycleManager.java:109)
    at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660)
    at com.datastax.kafkaconnector.state.LifeCycleManager.startTask(LifeCycleManager.java:105)
    at com.datastax.kafkaconnector.DseSinkTask.start(DseSinkTask.java:74)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.initializeAndStart(WorkerSinkTask.java:300)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:189)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
    Suppressed: com.datastax.oss.driver.api.core.connection.ConnectionInitException: [s1|control|id: 0xc4c37855, L:/172.21.117.41:52221 - R:test-db-cass0a02dc1/172.21.96.64:9042] Protocol initialization request, step 1 (OPTIONS): failed to send request (javax.net.ssl.SSLException: SSLEngine closed already)
        at com.datastax.oss.driver.internal.core.channel.ProtocolInitHandler$InitRequest.fail(ProtocolInitHandler.java:342)
        at com.datastax.oss.driver.internal.core.channel.ChannelHandlerRequest.writeListener(ChannelHandlerRequest.java:87)
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:500)
        at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:493)
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:472)
        at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:413)
        at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:538)
        at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:531)
        at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:111)
        at io.netty.util.internal.PromiseNotificationUtil.tryFailure(PromiseNotificationUtil.java:64)
        at io.netty.channel.DelegatingChannelPromiseNotifier.operationComplete(DelegatingChannelPromiseNotifier.java:57)
        at io.netty.channel.DelegatingChannelPromiseNotifier.operationComplete(DelegatingChannelPromiseNotifier.java:31)
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:500)
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:474)
        at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:413)
        at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:538)
        at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:531)
        at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:111)
        at io.netty.util.internal.PromiseNotificationUtil.tryFailure(PromiseNotificationUtil.java:64)
        at io.netty.channel.DelegatingChannelPromiseNotifier.operationComplete(DelegatingChannelPromiseNotifier.java:57)
        at io.netty.channel.DelegatingChannelPromiseNotifier.operationComplete(DelegatingChannelPromiseNotifier.java:31)
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:500)
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:474)
        at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:413)
        at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:538)
        at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:531)
        at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:111)
        at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:835)
        at io.netty.handler.ssl.SslHandler.wrapAndFlush(SslHandler.java:797)
        at io.netty.handler.ssl.SslHandler.handleUnwrapThrowable(SslHandler.java:1254)
        at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1230)
        at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1271)
        at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:505)
        at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:444)
        at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:283)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1421)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511)
        at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        ... 1 more
    Caused by: javax.net.ssl.SSLException: SSLEngine closed already
        at io.netty.handler.ssl.SslHandler.wrap(SslHandler.java:834)
        ... 23 more

Попробовал включить режим отладки для журналов SSl и обнаружил следующую ошибку: 

s0-io-0, fatal error: 46: General SSLEngine problem
java.security.cert.CertificateException: No name matching test-db-cass0a01dc1 found
%% Invalidated:  [Session-60, TLS_RSA_WITH_AES_256_CBC_SHA]
s0-io-0, SEND TLSv1.2 ALERT:  fatal, description = certificate_unknown
s0-io-0, WRITE: TLSv1.2 Alert, length = 2
s0-io-0, fatal: engine already closed.  Rethrowing javax.net.ssl.SSLHandshakeException: General SSLEngine problem
s0-io-0, called closeOutbound()
s0-io-0, closeOutboundInternal()
s0-io-0, called closeOutbound()
s0-io-0, closeOutboundInternal()
s0-io-0, called closeOutbound()
s0-io-0, closeOutboundInternal()
s0-io-0, called closeInbound()
s0-io-0, fatal: engine already closed.  Rethrowing javax.net.ssl.SSLException: Inbound closed before receiving peer's close_notify: possible truncation attack?
s0-io-0, called closeOutbound()
s0-io-0, closeOutboundInternal()
s0-io-0, called closeOutbound()
s0-io-0, closeOutboundInternal()
Using SSLEngineImpl.
Allow unsafe renegotiation: false
Allow legacy hello messages: true
Is initial handshake: true
Is secure renegotiation: false

test-db-cass0a01dc1 - это имя хоста для cassandra сервер с IP 172.21.96.64

1 Ответ

0 голосов
/ 28 апреля 2020

Вы можете добавить в раздел config следующий параметр, чтобы отключить проверку имени хоста: 

"datastax-java-driver.advanced.ssl-engine-factory.hostname-validation":"false"

См. Документацию Kafka Connector относительно передачи Java свойств драйвера и Java ссылка на конфигурацию драйвера .

...