Flume定制elasticsearch sink源码
最近尝试通过Flume将消息写入elasticsearch，但是flume并没有对每个es版本提供支持，仅仅保留了对0.9版本支持，可能是由于es版本变化频繁且不同版本间差异较大，没有办法在每个Flume版本都对es每个版本进行一次定制开发。
版本兼容问题
下面是我如何在flume1.7版本实现对es6.8写入，实现期间趟了无数的坑，其中一个下插曲是，自己大意，从官网下载了最新的flume源码（1.9），因为es sink部分代码变化极少，以因此窃以为使用最新源码开发只编译es sink包是没有问题的，开发完了才发现，打出的sink包无法在1.7上运行，重下1.7版本flume源码再做调整。。。ε=(´ο｀*)))唉。
Flume源码下载
flume是apache的顶级开源项目，直接到apache官网下载，源码下下来后使用IDE打开，我用的是Idea。flume有两个发版的代码线，0.9.x和1.x，这里要注意，下载的flume源码版本要和自己使用的Flume版本一致。flume项目依赖的包非常多，并且开源项目都使用的是Maven中心仓库里的官方包，因此第一次导入Flume项目是个很漫长的过程，保持网络畅通哦，我导入大概花了3个小时把所有包下下来。
代码修改
flume源码中es sink相关代码都在flume/flume-ng-sinks/flume-ng-elasticsearch-sink子模块下,代码实现很简单。
源码详见：https://github.com/liwutao/flume-with-es6.85
apache-flume-1.7.0-src
|—flume-ng-elasticsearch-sink
|—client
|—ElasticSearchClient.java
|—ElasticSearchClientFactory.java
|—ElasticSearchRestClient.java
|—ElasticSearchTransportClient.java
|—NoSuchClientTypeException.java
|—RoundRobinList.java
|—AbstractElasticSearchIndexRequestBuilderFactory.java
|—ContentBuilderUtil.java
|—ElasticSearchDynamicSerializer.java
|—ElasticSearchIndexRequestBuilderFactory.java
|—ElasticSearchLogStashEventSerializer.java
|—ElasticSearchSink.java
|—ElasticSearchSinkConstants.java
|—EventSerializerIndexRequestBuilderFactory.java
|—IndexNameBuilder.java
|—SimpleIndexNameBuilder.java
|—TimeBasedIndexNameBuilder.java
|—TimestampedEvent.java
|—pom.xml
|—pom.xml
1. 修改pom.xml 中es相关包依赖版本为6.8.5
2. 调整es sink代码中所有使用es接口的代码，都调整为使用6.8.5接口
3. 修改flume-ng-elasticsearch-sink子工程pom.xml，增加transport依赖，用于提供6.8.5客户端依赖
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>transport</artifactId>
</dependency>
4. 修改flume-ng-elasticsearch-sink子工程pom.xml，增加httpclient依赖，用于提供6.8.5客户端依赖
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
</dependency>
打包部署
修改完成后需要打包部署，将打出的flume-ng-elasticsearch-sink-1.7.0.jar 包部署到${FLUME_HOME}/lib/下
从es环境拷贝所有elastic相关包到${FLUME_HOME}/lib/下
从本地拷贝elasticsearch sink依赖的包到${FLUME_HOME}/lib/下， 依赖包挺多都是通过报错一一排查出来的：
elasticsearch-6.8.5.jar
elasticsearch-cli-6.8.5.jar
elasticsearch-core-6.8.5.jar
elasticsearch-rest-client-6.8.5.ja
elasticsearch-secure-sm-6.8.5.jar
elasticsearch-ssl-config-6.8.5.jar
elasticsearch-x-content-6.8.5.jar
httpasyncclient-4.1.2.jar
jackson-core-asl-1.9.3.jar.bak
lang-mustache-client-6.8.5.jar
netty-3.9.4.Final.jar
netty-buffer-4.1.32.Final.jar
netty-codec-4.1.32.Final.jar
netty-codec-http-4.1.32.Final.jar
netty-common-4.1.32.Final.jar
netty-handler-4.1.32.Final.jar
netty-resolver-4.1.32.Final.jar
netty-transport-4.1.32.Final.jar
parent-join-client-6.8.5.jar
percolator-client-6.8.5.jar
rank-eval-client-6.8.5.jar
reindex-client-6.8.5.jar
transport-6.8.5.jar
transport-netty4-client-6.8.5.jar
定制flume interceptor
趟坑
下面是遇到的几个缺包报错：
FAIL_ON_SYMBOL_HASH_OVERFLOW
11 三月 2020 12:16:31,586 ERROR [lifecycleSupervisor-1-2] (org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run:251) - Unable to start SinkRunner: { policy:org.apache.flume.sink.DefaultSinkProcessor@29ce66c0 counterGroup:{ name:null counters:{} } } - Exception follows. java.lang.NoSuchFieldError: FAIL_ON_SYMBOL_HASH_OVERFLOW at org.elasticsearch.common.xcontent.json.JsonXContent.<clinit>(JsonXContent.java:57) at org.elasticsearch.common.xcontent.XContentType$1.xContent(XContentType.java:56) at org.elasticsearch.common.settings.Setting.arrayToParsableString(Setting.java:1318) at org.elasticsearch.common.settings.Setting.access$800(Setting.java:87) at org.elasticsearch.common.settings.Setting$ListSetting.lambda$new$0(Setting.java:1343) at org.elasticsearch.common.settings.Setting$ListSetting.innerGetRaw(Setting.java:1353) at org.elasticsearch.common.settings.Setting.getRaw(Setting.java:461) at org.elasticsearch.common.settings.Setting.lambda$listSetting$35(Setting.java:1269) at org.elasticsearch.common.settings.Setting.listSetting(Setting.java:1286) at org.elasticsearch.common.settings.Setting.listSetting(Setting.java:1269) at org.elasticsearch.transport.TransportSettings.<clinit>(TransportSettings.java:47) at org.elasticsearch.client.transport.TransportClient.newPluginService(TransportClient.java:105) at org.elasticsearch.client.transport.TransportClient.buildTemplate(TransportClient.java:135) at org.elasticsearch.client.transport.TransportClient.<init>(TransportClient.java:288) at org.elasticsearch.transport.client.PreBuiltTransportClient.<init>(PreBuiltTransportClient.java:128) at org.elasticsearch.transport.client.PreBuiltTransportClient.<init>(PreBuiltTransportClient.java:114) at org.elasticsearch.transport.client.PreBuiltTransportClient.<init>(PreBuiltTransportClient.java:104) at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.openClient(ElasticSearchTransportClient.java:206) at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.<init>(ElasticSearchTransportClient.java:79) at org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory.getClient(ElasticSearchClientFactory.java:48) at org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:354) at org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:45) at org.apache.flume.SinkRunner.start(SinkRunner.java:79) at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:249) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) 11 三月 2020 12:16:31,590 INFO [lifecycleSupervisor-1-2] (org.apache.flume.sink.elasticsearch.ElasticSearchSink.stop:381) - ElasticSearch sink {} stopping
问题：所依赖jackson包版本不一致
解决：需要把本地打包使用的所有jackson包都替换到flume环境
ClassNotFound:io.netty.util.NettyRuntime
问题：缺少nettyCommon包
解决：把本地仓库netty目录下所有依赖包直接拷贝到flume环境
ClassNotFound:SslConfigurationLoader
问题：缺少elasticsearch-ssl-config包
解决方案：elasticsearch所有包都需要添加到flume
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch-ssl-config</artifactId>
<version>6.7.1</version>
</dependency>
ClassNotFound:SchemeIOSessionStrategy
unner: { policy:org.apache.flume.sink.DefaultSinkProcessor@6d310488 counterGroup:{ name:null counters:{} } } - Exception follows.
java.lang.NoClassDefFoundError: org/apache/http/nio/conn/SchemeIOSessionStrategy
at org.elasticsearch.index.reindex.ReindexPlugin.getSettings(ReindexPlugin.java:94)
at org.elasticsearch.plugins.PluginsService.lambda$getPluginSettings$0(PluginsService.java:89)
at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:267)
at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
at org.elasticsearch.plugins.PluginsService.getPluginSettings(PluginsService.java:89)
at org.elasticsearch.client.transport.TransportClient.buildTemplate(TransportClient.java:147)
at org.elasticsearch.client.transport.TransportClient.<init>(TransportClient.java:288)
at org.elasticsearch.transport.client.PreBuiltTransportClient.<init>(PreBuiltTransportClient.java:128)
at org.elasticsearch.transport.client.PreBuiltTransportClient.<init>(PreBuiltTransportClient.java:114)
at org.elasticsearch.transport.client.PreBuiltTransportClient.<init>(PreBuiltTransportClient.java:104)
at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.openClient(ElasticSearchTransportClient.java:206)
at org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.<init>(ElasticSearchTransportClient.java:79)
at org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory.getClient(ElasticSearchClientFactory.java:48)
at org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:354)
at org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:45)
at org.apache.flume.SinkRunner.start(SinkRunner.java:79)
at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:249)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException: org.apache.http.nio.conn.SchemeIOSessionStrategy
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 29 more
解决方案：httpasyncclient包需要拷贝到flume
java.lang.NoSuchMethodError: io.netty.util.internal.ObjectUtil.checkPositive(ILjava/lang/String;)
原因：netty-common包版本低或者版本冲突
解决：使用netty-common-4.1.32.Final.jar
两种客户端
flume elasticsearch sink访问es使用了两种客户端：
PreBuiltTransportClient
transportClient使用接口9300
HttpClient
restClient接口9200