自定义拦截器不能与Apache Flume配合使用

Question

我有一个Flume组件监听Syslog流。 我做了一个自定义拦截器来修改调用，但它不工作。我做错了什么？ 谢谢 安德烈

拦截器是一个精心编制的JAR Ile和在@ FLUME_HOME/bin中目录

拦截器类：

package com.test.flume; 

import org.apache.flume.Context; 
import org.apache.flume.Event; 
import org.apache.flume.conf.Configurable; 
import org.apache.flume.interceptor.Interceptor; 

import java.util.Iterator; 
import java.util.List; 
import java.util.concurrent.atomic.AtomicLong; 

public class SQLFlumeInterceptor implements Interceptor { 

    private final String headerKey; 

    private SQLFlumeInterceptor(Context ctx) {    
    } 

    @Override 
    public void initialize() { 

    } 

    @Override 
    public Event intercept(Event event) {    
     addPreposition(event); 
     return event; 
    } 

    private void addPreposition(Event event) {    
     System.out.println("Event processed"); 
     event.setBody("Modified Event".getBytes()); 
    } 

    @Override 
    public List<Event> intercept(List<Event> events) { 
     for (Iterator<Event> iterator = events.iterator(); iterator.hasNext();) { 

      Event next = iterator.next(); 
      intercept(next); 

      if(next == null) { 
      iterator.remove(); 
      } 
     } 
     return events; 
    } 

    @Override 
    public void close() { 

    } 

    public static class CounterInterceptorBuilder implements Interceptor.Builder { 

     private Context ctx; 

     @Override 
     public Interceptor build() { 
      return new SQLFlumeInterceptor(ctx); 
     } 

     @Override 
     public void configure(Context context) { 
     this.ctx = context; 
     } 

    } 

flume.config文件

# Name the components on this agent 
a1.sources = r1 
a1.sinks = file-sink 
a1.channels = c1 

# Describe/configure the source 
a1.sources.r1.type = syslogtcp 
a1.sources.r1.port = 41414 
a1.sources.r1.host = 192.168.1.2 
a1.sources.r1.interceptors = i1 
a1.sources.r1.interceptors.i1.type = com.test.flume.SQLFlumeInterceptor$CounterInterceptorBuilder 

# Describe the FILE_ROLLsink 
a1.sinks.file-sink.type = FILE_ROLL 
a1.sinks.file-sink.sink.directory = /opt/apache-flume-1.5.2-bin/logs/pluto.log 
a1.sinks.file-sink.sink.rollInterval = 0 
ai.sinks.file-sink.batchSize = 100 
ai.sinks.file-sink.fileHeader = true 


# Use a channel which buffers events in memory 
a1.channels.c1.type = memory 
a1.channels.c1.capacity = 1000 
a1.channels.c1.transactionCapacity = 100 

# Bind the source and sink to the channel 
a1.sources.r1.channels = c1 
a1.sinks.file-sink.channel = c1 

系统将事件记录在文件中而不修改它们，这是p ertinent DEBUG日志：

2015-04-27 21:39:17,625 (conf-file-poller-0) [DEBUG - org.apache.flume.conf.FlumeConfiguration$AgentConfiguration.isValid(FlumeConfiguration.java:313)] Starting validation of configuration for agent: a1, initial-configuration: AgentConfiguration[a1] 
SOURCES: {r1={ parameters:{port=41414, host=192.168.1.2, interceptors=i1, interceptors.i1.type=com.test.flume. 
SQLFlumeInterceptor$CounterInterceptorBuilder, channels=c1, type=syslogtcp} }} 
CHANNELS: {c1={ parameters:{transactionCapacity=100, capacity=1000, type=memory} }} 
SINKS: {file-sink={ parameters:{sink.rollInterval=0, type=FILE_ROLL, channel=c1, sink.directory=/opt/apache-flume-1.5.2-bin/logs/pluto.log} }} 

Answer 1

请放在@ FLUME_HOME/lib目录下的拦截JAR文件，而不是在@ FLUME_HOME /箱。

否则flume将不加载JAR。