[FLUME-2440] ElasticSearchIndexRequestBuilderFactory

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Not A Problem
• Affects Version/s: 1.5.0
• Component/s: Sinks+Sources
• Labels: patch

Environment

centos CDH

• Release Note: .
• Assignee: Unassigned
• Reporter: tu gongxuan
• Created: 04/Aug/14 02:20
• Updated: 11/Sep/14 23:20
• Resolved: 04/Aug/14 18:18

Description

I want sink the data to ElasticSearch with ElasticSearchIndexRequestBuilderFactory option, according to the document :
"The ElasticSearchIndexRequestBuilderFactory or ElasticSearchEventSerializer to use. Implementations of either class are accepted but ElasticSearchIndexRequestBuilderFactory is preferred."

but when start flume , error occurs. after check the source code, find the ElasticSearchIndexRequestBuilderFactory is just a interface.

my flume configure file as following:
agent1.sources=source1
agent1.sinks=sink1
agent1.channels=channel1

agent1.sources.source1.type=spooldir
agent1.sources.source1.spoolDir=/root/flume_test/spl
agent1.sources.source1.channels=channel1
agent1.sources.source1.fileHeader = false

agent1.sinks.sink1.channel=channel1
agent1.sinks.sink1.type=org.apache.flume.sink.elasticsearch.ElasticSearchSink
#agent1.sinks.sink1.type=elasticsearch
agent1.sinks.sink1.hostNames=192.168.106.27:9300
agent1.sinks.sink1.serializer=org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory
#agent1.sinks.sink1.serializer=org.apache.flume.sink.elasticsearch.ElasticSearchLogStashEventSerializer

agent1.channels.channel1.type=file
agent1.channels.channel1.checkpointDir=/root/flume_test/ck
agent1.channels.channel1.dataDirs=/root/flume_test/data

the error is :

14/08/04 10:04:38 INFO source.DefaultSourceFactory: Creating instance of source source1, type spooldir
14/08/04 10:04:38 INFO sink.DefaultSinkFactory: Creating instance of sink: sink1, type: org.apache.flume.sink.elasticsearch.ElasticSearchSink
14/08/04 10:04:38 ERROR elasticsearch.ElasticSearchSink: Could not instantiate event serializer.
java.lang.InstantiationException: org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory
at java.lang.Class.newInstance(Class.java:359)
at org.apache.flume.sink.elasticsearch.ElasticSearchSink.configure(ElasticSearchSink.java:295)
at org.apache.flume.conf.Configurables.configure(Configurables.java:41)
at org.apache.flume.node.AbstractConfigurationProvider.loadSinks(AbstractConfigurationProvider.java:418)
at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:103)
at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140)
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:744)
14/08/04 10:04:38 ERROR node.AbstractConfigurationProvider: Sink sink1 has been removed due to an error during configuration
java.lang.RuntimeException: java.lang.InstantiationException: org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory
at com.google.common.base.Throwables.propagate(Throwables.java:156)
at org.apache.flume.sink.elasticsearch.ElasticSearchSink.configure(ElasticSearchSink.java:309)
at org.apache.flume.conf.Configurables.configure(Configurables.java:41)
at org.apache.flume.node.AbstractConfigurationProvider.loadSinks(AbstractConfigurationProvider.java:418)
at org.apache.flume.node.AbstractConfigurationProvider.getConfiguration(AbstractConfigurationProvider.java:103)
at org.apache.flume.node.PollingPropertiesFileConfigurationProvider$FileWatcherRunnable.run(PollingPropertiesFileConfigurationProvider.java:140)
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:744)
Caused by: java.lang.InstantiationException: org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory
at java.lang.Class.newInstance(Class.java:359)
at org.apache.flume.sink.elasticsearch.ElasticSearchSink.configure(ElasticSearchSink.java:295)

pls help me !!!

thanks