[Issue 3275][pulsar-io]Support source and sink of flume

[tuteng]

Motivation

#3275

Modifications

• Support data of flume to pulsar
• Support data of pulsar to flume

Document

flow of data

sink command

pulsar/bin/pulsar-admin sink localrun --archive pulsar/pulsar-io/flume/target/pulsar-io-flume-2.3.0-SNAPSHOT.nar --classname org.apache.pulsar.io.flume.sink.StringSink --tenant public --namespace default --name flume  --inputs flume_test --sink-config-file pulsar/pulsar-io/flume/src/main/resources/flume/flume-io-source.yaml --parallelism 1

source command

pulsar/bin/pulsar-admin source localrun --archive pulsar/pulsar-io/flume/target/pulsar-io-flume-2.3.0-SNAPSHOT.nar --classname org.apache.pulsar.io.flume.source.StringSource --tenant public --namespace default --name flume  --destination-topic-name flume_test --source-config-file pulsar/pulsar-io/flume/src/main/resources/flume/flume-io-sink.yaml --parallelism 1

other document

In editing

[aahmed-se]

Flume has it's source sink system, not sure where we should integrate the source and sink

[tuteng]

Flume has it's source sink system, not sure where we should integrate the source and sink

@aahmed-se That's true. This question has been discussed in issue #3275.

[aahmed-se]

Sounds good.

[tuteng]

rerun java8 tests

[tuteng]

rerun java8 tests

[tuteng]

run java8 tests
run integration tests

[tuteng]

rerun java8 tests

[merlimat]

@tuteng I've seen a failure in the tests for this connector.

Standard Output
Created annotation listener
14:22:32.466 [main:org.apache.flume.source.AvroSource@205] ERROR org.apache.flume.source.AvroSource - Avro source null startup failed. Cannot initialize Netty server
org.jboss.netty.channel.ChannelException: Failed to bind to: /0.0.0.0:44444
	at org.jboss.netty.bootstrap.ServerBootstrap.bind(ServerBootstrap.java:272) ~[netty-3.10.1.Final.jar:?]
	at org.apache.avro.ipc.NettyServer.<init>(NettyServer.java:106) ~[avro-ipc-1.8.1.jar:1.8.1]
	at org.apache.flume.source.AvroSource.start(AvroSource.java:202) [flume-ng-core-1.9.0.jar:1.9.0]
	at org.apache.pulsar.io.flume.sink.StringSinkTests.setUp(StringSinkTests.java:82) [test-classes/:?]
	....
	at org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:413) [surefire-booter-2.21.0.jar:2.21.0]
Caused by: java.net.BindException: Address already in use

I was trying to see if it's possible to check for an open port and use that from the config files.

We have already a class to get that in:

 <dependency>
   <groupId>${project.groupId}</groupId>
   <artifactId>managed-ledger-original</artifactId>
   <version>${project.version}</version>
   <scope>test</scope>
</dependency>

The you could just call PortManager.nextFreePort().

[codelipenghui]

@tuteng Can you add some documents for the flume IO connector?

[tuteng]

@tuteng Can you add some documents for the flume IO connector?

Ok, I will add it later.