David Waugh created FLUME-2816:
----------------------------------
Summary: Unable to get apache-flume-1.6.0 to work with
ElasticSearch 2.0 beta 2 - Works with Elasticsearch 1.7.3
Key: FLUME-2816
URL: https://issues.apache.org/jira/browse/FLUME-2816
Project: Flume
Issue Type: Improvement
Components: Sinks+Sources
Affects Versions: v1.6.0
Environment: Centos 6
Reporter: David Waugh
Hello
I am unable to get apache-flume-1.6.0 to work with ElasticSearch 2.0 beta 2.
The same configuration works fine with Elasticsearch 1.7.3
My flume-en.sh contains
{noformat}
Give Flume more memory and pre-allocate, enable remote monitoring via JMX
export JAVA_OPTS="-Xms100m -Xmx2000m -Dcom.sun.management.jmxremote"
Note that the Flume conf directory is always included in the classpath.
FLUME_CLASSPATH="/usr/share/elasticsearch/lib/elasticsearch-2.0.0-beta2.jar"
export JAVA_HOME=/usr/lib/jvm/java-1.7.0-openjdk-1.7.0.85.x86_64/jre
{noformat}
I have copied the lucene-core-5.2.1.jar from /usr/share/elasticsearch/lib to
/opt/flume/lib
my elastic.conf file is:
elastic.conf
{noformat}
#elastic.conf: A single-node Flume configuration
#Reading avro files via spool dir and use the elastic sink
#Name the components on this agent
a1.sources = avrofld
a1.sinks = k1
a1.channels = c1
#Describe/configure the folder with the avro files from Warehouse Connector
a1.sources.avrofld.type = spooldir
a1.sources.avrofld.channels = c1
a1.sources.avrofld.spoolDir = /var/export
a1.sources.avrofld.fileHeader = true
a1.sources.avrofld.deserializer = avro
Describe the elasticsearch sink
a1.sinks.k1.type = org.apache.flume.sink.elasticsearch.ElasticSearchSink
a1.sinks.k1.hostNames = 127.0.0.1:9300
a1.sinks.k1.indexName = sa
a1.sinks.k1.indexType = data
a1.sinks.k1.clusterName = elasticsearch
a1.sinks.k1.batchSize = 1000
a1.sinks.k1.ttl = 2d
a1.sinks.k1.serializer = com.rsa.flume.serialization.FlumeAvroEventDeserializer
#Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000000
a1.channels.c1.transactionCapacity = 100000
#Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
{noformat}
Elasticsearch 2.0 is successfully started and listening on the port
{noformat}
[root@CentosB lib]# netstat -na |grep 900
tcp 0 0 127.0.0.1:9200 0.0.0.0: LISTEN
tcp 0 0 127.0.0.1:9300 0.0.0.0:* LISTEN
{noformat}
However when starting my flume I get the following error
{noformat}
[root@CentosB flume]# bin/flume-ng agent --conf conf --conf-file
conf/elastic.conf --name a1 -Dflume.root.logger=DEBUG,console
2015-10-19 10:24:31,152 (lifecycleSupervisor-1-1) [ERROR -
org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:253)]
Unable to start SinkRunner: {
policy:org.apache.flume.sink.DefaultSinkProcessor@5c5dc0a5 counterGroup:{
name:null counters:{} } } - Exception follows.
java.lang.NoSuchMethodError:
org.elasticsearch.common.transport.InetSocketTransportAddress.(Ljava/lang/String;I)V
at
org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.configureHostnames(ElasticSearchTransportClient.java:143)
at
org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.(ElasticSearchTransportClient.java:77)
at
org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory.getClient(ElasticSearchClientFactory.java:48)
at
org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:357)
at
org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:46)
at org.apache.flume.SinkRunner.start(SinkRunner.java:79)
at
org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
{noformat}
Full Log Below
{noformat}
Info: Sourcing environment configuration script /opt/flume/conf/flume-env.sh
Info: Including Hive libraries found via () for Hive access
exec /usr/lib/jvm/java-1.7.0-openjdk-1.7.0.85.x86_64/jre/bin/java -Xms100m
-Xmx2000m -Dcom.sun.management.jmxremote -Dflume.root.logger=DEBUG,console -cp
'/opt/flume/conf:/opt/flume/lib/:/usr/share/elasticsearch/lib/elasticsearch-2.0.0-beta2.jar:/lib/'
-Djava.library.path= org.apache.flume.node.Application --conf-file
conf/elastic.conf --name a1 2015-10-19 10:24:31,017 (lifecycleSupervisor-1-0)
[INFO -
org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:61)]
Configuration provider starting 2015-10-19 10:24:31,020
(lifecycleSupervisor-1-0) [DEBUG -
org.apache.flume.node.PollingPropertiesFileConfigurationProvider.start(PollingPropertiesFileConfigurationProvider.java:78)]
Configuration provider started 2015-10-19 10:24:31,023 (conf-file-poller-0)
[DEBUG -
org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:126)]
Checking file:conf/elastic.conf for changes 2015-10-19 10:24:31,023
(conf-file-poller-0) [INFO -
org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:133)]
Reloading configuration file:conf/elastic.conf 2015-10-19 10:24:31,028
(conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,028 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1021)]
Created context for k1: serializer 2015-10-19 10:24:31,029
(conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,029 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:931)]
Added sinks: k1 Agent: a1 2015-10-19 10:24:31,029 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,029 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,029 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,029 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,030 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,030 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,030 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.addProperty(FlumeConfiguration.java:1017)]
Processing:k1 2015-10-19 10:24:31,030 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:314)]
Starting validation of configuration for agent: a1, initial-configuration:
AgentConfiguration[a1] SOURCES: {avrofld={ parameters:{fileHeader=true,
channels=c1, spoolDir=/var/export, type=spooldir, deserializer=avro} }, r1={
parameters:{channels=c1} }} CHANNELS: {c1={
parameters:{transactionCapacity=100000, capacity=1000000, type=memory} }}
SINKS: {k1={ parameters:{clusterName=elasticsearch, indexType=data,
serializer=com.rsa.flume.serialization.FlumeAvroEventDeserializer,
indexName=sa, batchSize=1000, hostNames=127.0.0.1:9300,
type=org.apache.flume.sink.elasticsearch.ElasticSearchSink, ttl=2d, channel=c1}
}} 2015-10-19 10:24:31,039 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateChannels(FlumeConfiguration.java:469)]
Created channel c1 2015-10-19 10:24:31,046 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.validateSinks(FlumeConfiguration.java:675)]
Creating sink: k1 using ELASTICSEARCH 2015-10-19 10:24:31,047
(conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:372)]
Post validation configuration for a1 AgentConfiguration created without
Configuration stubs for which only basic syntactical validation was
performed[a1] SOURCES: {avrofld={ parameters:{fileHeader=true, channels=c1,
spoolDir=/var/export, type=spooldir, deserializer=avro} }} CHANNELS: {c1={
parameters:{transactionCapacity=100000, capacity=1000000, type=memory} }}
SINKS: {k1={ parameters:{clusterName=elasticsearch, indexType=data,
serializer=com.rsa.flume.serialization.FlumeAvroEventDeserializer,
indexName=sa, batchSize=1000, hostNames=127.0.0.1:9300,
type=org.apache.flume.sink.elasticsearch.ElasticSearchSink, ttl=2d, channel=c1}
}} 2015-10-19 10:24:31,047 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:136)]
Channels:c1 2015-10-19 10:24:31,048 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:137)]
Sinks k1 2015-10-19 10:24:31,048 (conf-file-poller-0) [DEBUG -
org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:138)]
Sources avrofld 2015-10-19 10:24:31,048 (conf-file-poller-0) [INFO -
org.apache.flume.conf.FlumeConfiguration.validateConfiguration(FlumeConfiguration.java:141)]
Post-validation flume configuration contains configuration for agents: [a1]
2015-10-19 10:24:31,048 (conf-file-poller-0) [INFO -
org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:145)]
Creating channels 2015-10-19 10:24:31,056 (conf-file-poller-0) [INFO -
org.apache.flume.channel.DefaultChannelFactory.create(DefaultChannelFactory.java:42)]
Creating instance of channel c1 type memory 2015-10-19 10:24:31,063
(conf-file-poller-0) [INFO -
org.apache.flume.node.AbstractConfigurationProvider.loadChannels(AbstractConfigurationProvider.java:200)]
Created channel c1 2015-10-19 10:24:31,063 (conf-file-poller-0) [INFO -
org.apache.flume.source.DefaultSourceFactory.create(DefaultSourceFactory.java:41)]
Creating instance of source avrofld, type spooldir 2015-10-19 10:24:31,086
(conf-file-poller-0) [INFO -
org.apache.flume.sink.DefaultSinkFactory.create(DefaultSinkFactory.java:42)]
Creating instance of sink: k1, type:
org.apache.flume.sink.elasticsearch.ElasticSearchSink 2015-10-19 10:24:31,086
(conf-file-poller-0) [DEBUG -
org.apache.flume.sink.DefaultSinkFactory.getClass(DefaultSinkFactory.java:63)]
Sink type org.apache.flume.sink.elasticsearch.ElasticSearchSink is a custom
type 2015-10-19 10:24:31,115 (conf-file-poller-0) [INFO -
org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:114)]
Channel c1 connected to [avrofld, k1] 2015-10-19 10:24:31,124
(conf-file-poller-0) [INFO -
org.apache.flume.node.Application.startAllComponents(Application.java:138)]
Starting new configuration:{ sourceRunners:{avrofld=EventDrivenSourceRunner: {
source:Spool Directory source avrofld: { spoolDir: /var/export } }}
sinkRunners:{k1=SinkRunner: {
policy:org.apache.flume.sink.DefaultSinkProcessor@5c5dc0a5 counterGroup:{
name:null counters:{} } }}
channels:{c1=org.apache.flume.channel.MemoryChannel{name: c1}} } 2015-10-19
10:24:31,134 (conf-file-poller-0) [INFO -
org.apache.flume.node.Application.startAllComponents(Application.java:145)]
Starting Channel c1 2015-10-19 10:24:31,136 (lifecycleSupervisor-1-0) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)]
Monitored counter group for type: CHANNEL, name: c1: Successfully registered
new MBean. 2015-10-19 10:24:31,136 (lifecycleSupervisor-1-0) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)]
Component type: CHANNEL, name: c1 started 2015-10-19 10:24:31,137
(conf-file-poller-0) [INFO -
org.apache.flume.node.Application.startAllComponents(Application.java:173)]
Starting Sink k1 2015-10-19 10:24:31,138 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:350)]
ElasticSearch sink {} started 2015-10-19 10:24:31,139
(lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)]
Monitored counter group for type: SINK, name: k1: Successfully registered new
MBean. 2015-10-19 10:24:31,139 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)]
Component type: SINK, name: k1 started 2015-10-19 10:24:31,142
(conf-file-poller-0) [INFO -
org.apache.flume.node.Application.startAllComponents(Application.java:184)]
Starting Source avrofld 2015-10-19 10:24:31,142 (lifecycleSupervisor-1-2) [INFO
-
org.apache.flume.source.SpoolDirectorySource.start(SpoolDirectorySource.java:78)]
SpoolDirectorySource source starting with directory: /var/export 2015-10-19
10:24:31,147 (lifecycleSupervisor-1-1) [WARN -
org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.configureHostnames(ElasticSearchTransportClient.java:136)]
[127.0.0.1:9300] 2015-10-19 10:24:31,152 (lifecycleSupervisor-1-1) [ERROR -
org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:253)]
Unable to start SinkRunner: {
policy:org.apache.flume.sink.DefaultSinkProcessor@5c5dc0a5 counterGroup:{
name:null counters:{} } } - Exception follows. java.lang.NoSuchMethodError:
org.elasticsearch.common.transport.InetSocketTransportAddress.(Ljava/lang/String;I)V
at
org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.configureHostnames(ElasticSearchTransportClient.java:143)
at
org.apache.flume.sink.elasticsearch.client.ElasticSearchTransportClient.(ElasticSearchTransportClient.java:77)
at
org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory.getClient(ElasticSearchClientFactory.java:48)
at
org.apache.flume.sink.elasticsearch.ElasticSearchSink.start(ElasticSearchSink.java:357)
at
org.apache.flume.sink.DefaultSinkProcessor.start(DefaultSinkProcessor.java:46)
at org.apache.flume.SinkRunner.start(SinkRunner.java:79) at
org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471) at
java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304) at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:178)
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745) 2015-10-19 10:24:31,160
(lifecycleSupervisor-1-1) [INFO -
org.apache.flume.sink.elasticsearch.ElasticSearchSink.stop(ElasticSearchSink.java:376)]
ElasticSearch sink {} stopping 2015-10-19 10:24:31,160
(lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:150)]
Component type: SINK, name: k1 stopped 2015-10-19 10:24:31,160
(lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:156)]
Shutdown Metric for type: SINK, name: k1. sink.start.time == 1445246671139
2015-10-19 10:24:31,160 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:162)]
Shutdown Metric for type: SINK, name: k1. sink.stop.time == 1445246671160
2015-10-19 10:24:31,160 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.batch.complete == 0 2015-10-19
10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.batch.empty == 0 2015-10-19
10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.batch.underflow == 0 2015-10-19
10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.connection.closed.count == 1
2015-10-19 10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.connection.creation.count == 0
2015-10-19 10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.connection.failed.count == 0
2015-10-19 10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.event.drain.attempt == 0
2015-10-19 10:24:31,161 (lifecycleSupervisor-1-1) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.stop(MonitoredCounterGroup.java:178)]
Shutdown Metric for type: SINK, name: k1. sink.event.drain.sucess == 0
2015-10-19 10:24:31,161 (lifecycleSupervisor-1-1) [WARN -
org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:260)]
Component SinkRunner: {
policy:org.apache.flume.sink.DefaultSinkProcessor@5c5dc0a5 counterGroup:{
name:null counters:{} } } stopped, since it could not besuccessfully started
due to missing dependencies 2015-10-19 10:24:31,155 (lifecycleSupervisor-1-2)
[DEBUG -
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.(ReliableSpoolingFileEventReader.java:138)]
Initializing ReliableSpoolingFileEventReader with directory=/var/export,
metaDir=.flumespool, deserializer=avro 2015-10-19 10:24:31,175
(lifecycleSupervisor-1-2) [DEBUG -
org.apache.flume.client.avro.ReliableSpoolingFileEventReader.(ReliableSpoolingFileEventReader.java:160)]
Successfully created and deleted canary file:
/var/export/flume-spooldir-perm-check-8040653594475164987.canary 2015-10-19
10:24:31,176 (lifecycleSupervisor-1-2) [DEBUG -
org.apache.flume.source.SpoolDirectorySource.start(SpoolDirectorySource.java:111)]
SpoolDirectorySource source started 2015-10-19 10:24:31,177
(lifecycleSupervisor-1-2) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.register(MonitoredCounterGroup.java:120)]
Monitored counter group for type: SOURCE, name: avrofld: Successfully
registered new MBean. 2015-10-19 10:24:31,177 (lifecycleSupervisor-1-2) [INFO -
org.apache.flume.instrumentation.MonitoredCounterGroup.start(MonitoredCounterGroup.java:96)]
Component type: SOURCE, name: avrofld started
{noformat}
I opened a gihub issue
Looks like it is due to this change
https://www.elastic.co/guide/en/elasticsearch/reference/2.0/_java_api_changes.html#_inetsockettransportaddress_removed
And this looks to be because the ElasticSerach API has changed.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
