You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flume.apache.org by "David Waugh (JIRA)" <ji...@apache.org> on 2015/10/20 13:56:27 UTC
[jira] [Created] (FLUME-2816) Unable to get apache-flume-1.6.0 to
work with ElasticSearch 2.0 beta 2 - Works with Elasticsearch 1.7.3
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)