如何将Kafka的Elasticearch与SSL连接起来？

2022-11-29 10:00:36 other 阅读：639
本文介绍了如何将Kafka的Elasticearch与SSL连接起来？的处理方法，对大家解决问题具有一定的参考价值

问题描述

Hi查看https://docs.confluent.io/current/connect/kafka-connect-elasticsearch/configuration_options.html#security后，我不确定需要哪个配置才能将Kafka连接到位于https的Elasticearch终结点。

我可以将Kafka连接到不同的非HTTPS Elasticearch，并且在常规客户端上，我可以通过关闭SSL验证连接到HTTPS。

这个来自不同服务提供商的示例https://help.aiven.io/en/articles/2349675-aiven-kafka-elasticsearch-sink-connector使connection.url、connection.username和connection.password对于HTTPS来说似乎足够了，但我收到错误：

org.apache.kafka.connect.errors.ConnectException: Couldn't start ElasticsearchSinkTask due to connection error:
    at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:132)
    at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:122)
    at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.start(ElasticsearchSinkTask.java:122)
    at io.confluent.connect.elasticsearch.ElasticsearchSinkTask.start(ElasticsearchSinkTask.java:51)
    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:175)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
    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)

Caused by: javax.net.ssl.SSLHandshakeException: 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.ssl.Alerts.getSSLException(Alerts.java:192)
    at sun.security.ssl.SSLSocketImpl.fatal(SSLSocketImpl.java:1964)
    at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:328)
    at sun.security.ssl.Handshaker.fatalSE(Handshaker.java:322)
    at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1614)
    at sun.security.ssl.ClientHandshaker.processMessage(ClientHandshaker.java:216)
    at sun.security.ssl.Handshaker.processLoop(Handshaker.java:1052)
    at sun.security.ssl.Handshaker.process_record(Handshaker.java:987)
    at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:1072)
    at sun.security.ssl.SSLSocketImpl.performInitialHandshake(SSLSocketImpl.java:1385)
    at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1413)
    at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1397)
    at org.apache.http.conn.ssl.SSLConnectionSocketFactory.createLayeredSocket(SSLConnectionSocketFactory.java:394)
    at org.apache.http.conn.ssl.SSLConnectionSocketFactory.connectSocket(SSLConnectionSocketFactory.java:353)
    at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:141)
    at org.apache.http.impl.conn.PoolingHttpClientConnectionManager.connect(PoolingHttpClientConnectionManager.java:353)
    at org.apache.http.impl.execchain.MainClientExec.establishRoute(MainClientExec.java:380)
    at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:236)
    at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:184)
    at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:88)
    at org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110)
    at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:184)
    at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:82)
    at io.searchbox.client.http.JestHttpClient.executeRequest(JestHttpClient.java:115)
    at io.searchbox.client.http.JestHttpClient.execute(JestHttpClient.java:57)
    at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.getServerVersion(JestElasticsearchClient.java:206)
    at io.confluent.connect.elasticsearch.jest.JestElasticsearchClient.<init>(JestElasticsearchClient.java:130)
    ... 12 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:397)
    at sun.security.validator.PKIXValidator.engineValidate(PKIXValidator.java:302)
    at sun.security.validator.Validator.validate(Validator.java:260)
    at sun.security.ssl.X509TrustManagerImpl.validate(X509TrustManagerImpl.java:324)
    at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:229)
    at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:124)
    at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1596)
    ... 34 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:280)
    at sun.security.validator.PKIXValidator.doBuild(PKIXValidator.java:392)
    ... 40 more

在本地运行Confluent 5.2.2，Elasticearch 6.8.3。

配置：

{
  "name": "sink-elasticsearch-mytable-ssl",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://localhost:8081",
    "value.converter.schema.registry.url": "http://localhost:8081",

    "connection.url": "https://xxx.databases.appdomain.cloud:30810",
    "connection.username": "ibm_cloud_xxx",
    "connection.password": "xxx",

    "type.name": "type.name=kafka-connect",
    "topics": "postgresql-mytable",
    "key.ignore": true
  }
}

我使用的是托管Elasticearch服务，所以我得到的只是用户名/密码和TLS证书，所以我看不到密钥库/信任库之类的东西。要将Kafka Connect Elasticearch与SSL连接起来，我还需要做些什么？

推荐答案

如果使用弹性云，只需传递用户名和密码：

{
  "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
  "connection.url": "https://myclusterid.us-east-1.aws.found.io:9243",
  "connection.username": "elastic",
  "connection.password": "covfefe",
  […]

这篇关于如何将Kafka的Elasticearch与SSL连接起来？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，WP2

admin_action_{$_REQUEST[‘action’]}

do_action( "admin_action_{$_REQUEST[‘action’]}" )动作钩子：：在发送“Action”请求变量时激发。Action Hook: Fires when an ‘action’ request variable is sent.目录锚点：#说明#源码说明（Description）钩子名称的动态部分$_REQUEST['action']引用从GET或POST请求派生的操作。源码（Source）更新版本源码位置使用被使用2.6.0 wp-admin/admin.php:...

日期：2020-09-02 17:44:16 浏览：1172

admin_footer-{$GLOBALS[‘hook_suffix’]}

do_action( "admin_footer-{$GLOBALS[‘hook_suffix’]}", string $hook_suffix )操作挂钩：在默认页脚脚本之后打印脚本或数据。Action Hook: Print scripts or data after the default footer scripts.目录锚点：#说明#参数#源码说明（Description）钩子名的动态部分，$GLOBALS['hook_suffix']引用当前页的全局钩子后缀。参数（Parameters）参数类...

日期：2020-09-02 17:44:20 浏览：1071

customize_save_{$this->id_data[‘base’]}

do_action( "customize_save_{$this-&gt;id_data[‘base’]}", WP_Customize_Setting $this )动作钩子：：在调用WP_Customize_Setting：：save（）方法时激发。Action Hook: Fires when the WP_Customize_Setting::save() method is called.目录锚点：#说明#参数#源码说明（Description）钩子名称的动态部分，$this->id_data...

日期：2020-08-15 15:47:24 浏览：808

customize_value_{$this->id_data[‘base’]}

apply_filters( "customize_value_{$this-&gt;id_data[‘base’]}", mixed $default )过滤器：：过滤未作为主题模式或选项处理的自定义设置值。Filter Hook: Filter a Customize setting value not handled as a theme_mod or option.目录锚点：#说明#参数#源码说明（Description）钩子名称的动态部分，$this->id_date['base']，指的是设置...

日期：2020-08-15 15:47:24 浏览：900

get_comment_author_url

过滤钩子：过滤评论作者的URL。Filter Hook: Filters the comment author’s URL.目录锚点：#源码源码（Source）更新版本源码位置使用被使用 wp-includes/comment-template.php:32610...

日期：2020-08-10 23:06:14 浏览：930

network_admin_edit_{$_GET[‘action’]}

do_action( "network_admin_edit_{$_GET[‘action’]}" )操作挂钩：启动请求的处理程序操作。Action Hook: Fires the requested handler action.目录锚点：#说明#源码说明（Description）钩子名称的动态部分$u GET['action']引用请求的操作的名称。源码（Source）更新版本源码位置使用被使用3.1.0 wp-admin/network/edit.php:3600...

日期：2020-08-02 09:56:09 浏览：877

network_sites_updated_message_{$_GET[‘updated’]}

apply_filters( "network_sites_updated_message_{$_GET[‘updated’]}", string $msg )筛选器挂钩：在网络管理中筛选特定的非默认站点更新消息。Filter Hook: Filters a specific, non-default site-updated message in the Network admin.目录锚点：#说明#参数#源码说明（Description）钩子名称的动态部分$_GET['updated']引用了非默认的...

日期：2020-08-02 09:56:03 浏览：865

pre_wp_is_site_initialized

过滤器：：过滤在访问数据库之前是否初始化站点的检查。Filter Hook: Filters the check for whether a site is initialized before the database is accessed.目录锚点：#源码源码（Source）更新版本源码位置使用被使用 wp-includes/ms-site.php:93910...

日期：2020-07-29 10:15:38 浏览：834

WordPress 的SEO 教学：如何在网站中加入关键字（Meta Keywords）与Meta 描述（Meta Description）？

你想在WordPress 中添加关键字和meta 描述吗？关键字和meta 描述使你能够提高网站的SEO。在本文中，我们将向你展示如何在WordPress 中正确添加关键字和meta 描述。为什么要在WordPress 中添加关键字和Meta 描述？关键字和说明让搜寻引擎更了解您的帖子和页面的内容。关键词是人们寻找您发布的内容时，可能会搜索的重要词语或片语。而Meta Description则是对你的页面和文章的简要描述。如果你想要了解更多关于中继标签的资讯，可以参考Google的说明。Meta 关键字和描...

日期：2020-10-03 21:18:25 浏览：1733

谷歌的SEO是什么

SEO (Search Engine Optimization)中文是搜寻引擎最佳化，意思近于「关键字自然排序」、「网站排名优化」。简言之，SEO是以搜索引擎（如Google、Bing）为曝光媒体的行销手法。例如搜寻「wordpress教学」，会看到本站的「WordPress教学：12个课程…」排行Google第一：关键字：wordpress教学、wordpress课程…若搜寻「网站架设」，则会看到另一个网页排名第1：关键字：网站架设、架站…以上两个网页，每月从搜寻引擎导入自然流量，达2万4千：每月「有机搜...

日期：2020-10-30 17:23:57 浏览：1309