水槽代理未在不同的机器上连接

Question

水槽代理1没有连接到水槽代理2。原因可能是什么？

我正在使用Flume使用2代理将日志文件流式传输到HDFS。第一个代理位于存在日志文件的源机器上，而第二个代理位于安装Hadoop的机器上（IP地址10.10.201.40）。

第一个代理（flume-src-agent.conf）的配置文件如下：

source_agent.sources = weblogic_server
source_agent.sources.weblogic_server.type = exec
source_agent.sources.weblogic_server.command = tail -f AdminServer.log
source_agent.sources.weblogic_server.batchSize = 1
source_agent.sources.weblogic_server.channels = memoryChannel
source_agent.sources.weblogic_server.interceptors = itime ihost itype

source_agent.sources.weblogic_server.interceptors.itime.type = timestamp

source_agent.sources.weblogic_server.interceptors.ihost.type = host
source_agent.sources.weblogic_server.interceptors.ihost.useIP = false
source_agent.sources.weblogic_server.interceptors.ihost.hostHeader = host

source_agent.sources.weblogic_server.interceptors.itype.type = static
source_agent.sources.weblogic_server.interceptors.itype.key = log_type
source_agent.sources.weblogic_server.interceptors.itype.value = apache_access_combined

source_agent.channels = memoryChannel
source_agent.channels.memoryChannel.type = memory
source_agent.channels.memoryChannel.capacity = 100


source_agent.sinks = avro_sink
source_agent.sinks.avro_sink.type = avro
source_agent.sinks.avro_sink.channel = memoryChannel
source_agent.sinks.avro_sink.hostname = 10.10.201.40

source_agent.sinks.avro_sink.port = 4545

第二个代理（flume-trg-agent.conf）的配置文件如下：

collector.sources = AvroIn
collector.sources.AvroIn.type = avro
collector.sources.AvroIn.bind = 0.0.0.0
collector.sources.AvroIn.port = 4545
collector.sources.AvroIn.channels = mc1 mc2

collector.channels = mc1 mc2
collector.channels.mc1.type = memory
collector.channels.mc1.capacity = 100
collector.channels.mc2.type = memory
collector.channels.mc2.capacity = 100

collector.sinks = HadoopOut
collector.sinks.HadoopOut.type = hdfs
collector.sinks.HadoopOut.channel = mc2
collector.sinks.HadoopOut.hdfs.path = hdfs://localhost:54310/user/root
collector.sinks.HadoopOut.hdfs.callTimeout = 150000
collector.sinks.HadoopOut.hdfs.fileType = DataStream
collector.sinks.HadoopOut.hdfs.writeFormat = Text
collector.sinks.HadoopOut.hdfs.rollSize = 0
collector.sinks.HadoopOut.hdfs.rollCount = 10000
collector.sinks.HadoopOut.hdfs.rollInterval = 600

当第一个代理运行时，我收到以下错误：

2015-04-08 15:14:10,251 (SinkRunner-PollingRunner-DefaultSinkProcessor) [ERROR - org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:160)] Unable to deliver event. Exception follows.
org.apache.flume.EventDeliveryException:Failed to send events

at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:382)

at org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)

at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)

at java.lang.Thread.run(Thread.java:744)

Caused by: org.apache.flume.FlumeException: NettyAvroRpcClient {host:10.10.201.40, port:4545}: RPC connection error

at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:161)

at org.apache.flume.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:115)

at org.apache.flume.api.NettyAvroRpcClient.configure(NettyAvroRpcClient.java:590)

at org.apache.flume.api.RpcClientFactory.getInstance(RpcClientFactory.java:88)

at org.apache.flume.sink.AvroSink.initializeRpcClient(AvroSink.java:127)

at org.apache.flume.sink.AbstractRpcSink.createConnection(AbstractRpcSink.java:209)

at org.apache.flume.sink.AbstractRpcSink.verifyConnection(AbstractRpcSink.java:269)

at org.apache.flume.sink.AbstractRpcSink.process(AbstractRpcSink.java:339)

... 3 more

Caused by: java.io.IOException: Error connecting to /10.10.201.40:4545

at org.apache.avro.ipc.NettyTransceiver.getChannel(NettyTransceiver.java:261)

at.org.apache.avro.ipc.NettyTransceiver.<init>(NettyTransceiver.java:203)

at.org.apache.avro.ipc.NettyTransceiver.<init>(NettyTransceiver.java:152)

at.org.apache.avro.api.NettyAvroRpcClient.connect(NettyAvroRpcClient.java:147)

运行第二个代理时，出现以下错误：

2015-04-08 15:53:31,649 (SinkRunner-PollingRunner-DefaultSinkProcessor) [DEBUG-org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:143)] Pollingsink runner starting

2015-04-08 15:53:31,844 (lifecycleSupervisor-1-3) [ERROR - org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:253)] Unable to start EventDrivenSourceRunner: { source:Avro source AvroIn: {bindAddress: 0.0.0.0, port: 4545 } } - Exception follows.

org.jboss.netty.channel.ChannelException: Failed to bind to: /0.0.0.0:4545

    at org.jboss.netty.bootstrap.ServerBootstrap.bind(ServerBootstrap.java:298)

    at org.apache.avro.ipc.NettyServer.<init>(NettyServer.java:106)

    at org.apache.flume.source.AvroSource.start(AvroSource.java:225)

    at org.apache.flume.source.EventDrivenSourceRunner.start(EventDrivenSourceRunner.java:44)

    at org.apache.flume.lifecycle.LifecycleSupervisor$MonitorRunnable.run(LifecycleSupervisor.java:251)

    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:441)

    at java.util.concurrent.FutureTask$Sync.innerRunAndReset(FutureTask.java:317)

    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:150)

    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$101(ScheduledThreadPoolExecutor.java:98)

    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.runPeriodic(ScheduledThreadPoolExecutor.java:180)

    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:204)

    at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)

    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)

    at java.lang.Thread.run(Thread.java:662)
Caused by: java.net.BindException: Address already in use

    at sun.nio.ch.Net.bind(Native Method)

    at sun.nio.ch.ServerSocketChannelImpl.bind(ServerSocketChannelImpl.java:126)

    at sun.nio.ch.ServerSocketAdaptor.bind(ServerSocketAdaptor.java:59)

    at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.bind(NioServerSocketPipelineSink.java:138)

    at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.handleServerSocket(NioServerSocketPipelineSink.java:90)

    at org.jboss.netty.channel.socket.nio.NioServerSocketPipelineSink.eventSunk(NioServerSocketPipelineSink.java:64)

    at org.jboss.netty.channel.Channels.bind(Channels.java:569)

    at org.jboss.netty.channel.AbstractChannel.bind(AbstractChannel.java:187)

    at org.jboss.netty.bootstrap.ServerBootstrap$Binder.channelOpen(ServerBootstrap.java:343)

    at org.jboss.netty.channel.Channels.fireChannelOpen(Channels.java:170)  

    at org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory.newChannel(NioServerSocketChannelFactory.java:158)

    at org.jboss.netty.channel.socket.nio.NioServerSocketChannel.<init>(NioServerSocketChannel.java:80)

    at org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory.newChannel(NioServerSocketChannelFactory.java:86)

    at org.jboss.netty.bootstrap.ServerBootstrap.bind(ServerBootstrap.java:277)
    ... 13 more

Answer 1

对于绑定问题，请键入netstat-plten并检查进程的pid并终止进程。这样做将在您再次运行代理时解决绑定问题

Answer 2

你的问题的答案在第二个日志里:

地址已被使用

这是因为有另一个进程正在使用端口4545。只需将两个代理重新配置到另一个端口，比如41414，就应该可以工作了。