[Issue 3275][pulsar-io]Support source and sink of flume (apache#3597)

* Support TLS authentication and authorization in standalone mode

* Compile success
To do: test channel, sink and source of flume

* Add conf file

* Add sink and source folder
Move file to folder
Add source
Compile success
To do -> test source

* test flume source paas

* Add config file and test case

* Add test and update pom.xml
To do add test of source and sink

* Add unit tests

* Add test case and test pass
To do test source

* Add license
Add test source of pulsar

* Handle if blockingQueue is null

* Move LOG to log

* Format code

* Add sinkClass in pulsar-io.yaml

* Add comment

* Default is false

* Modify pom file of flume

* Format pom.xml file

* Move pom version to 2.4.0-SNAPSHOT

pulsar-io/flume/pom.xml

@@ -0,0 +1,116 @@
+<!--
+
+    Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.pulsar</groupId>
+        <artifactId>pulsar-io</artifactId>
+        <version>2.4.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>pulsar-io-flume</artifactId>
+    <name>Pulsar IO :: Flume</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>pulsar-io-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-yaml</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flume</groupId>
+            <artifactId>flume-ng-node</artifactId>
+            <version>1.9.0</version>
+            <type>pom</type>
+            <exclusions>
+                <exclusion>
+                    <artifactId>avro-ipc</artifactId>
+                    <groupId>org.apache.avro</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>avro</artifactId>
+                    <groupId>org.apache.avro</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>1.8.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro-ipc</artifactId>
+            <version>1.8.1</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+            <version>4.10</version>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-all</artifactId>
+            <scope>test</scope>
+            <version>1.9.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>2.6.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.github.stefanbirkner</groupId>
+            <artifactId>system-rules</artifactId>
+            <scope>test</scope>
+            <version>1.17.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>18.0</version>
+        </dependency>
+    </dependencies>
+
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.nifi</groupId>
+                <artifactId>nifi-nar-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/FlumeConfig.java

@@ -0,0 +1,77 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.flume;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import lombok.*;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ *  Flume general config.
+ */
+@Data
+@Setter
+@Getter
+@EqualsAndHashCode
+@ToString
+@Accessors(chain = true)
+public class FlumeConfig {
+
+    @FieldDoc(
+            required = true,
+            defaultValue = "",
+            help = "the name of this agent")
+    private String name;
+    @FieldDoc(
+            required = true,
+            defaultValue = "",
+            help = "specify a config file (required if -z missing)")
+    private String confFile;
+    @FieldDoc(
+            defaultValue = "false",
+            help = "do not reload config file if changed")
+    private Boolean noReloadConf;
+    @FieldDoc(
+            required = true,
+            defaultValue = "",
+            help = "specify the ZooKeeper connection to use (required if -f missing)")
+    private String zkConnString;
+    @FieldDoc(
+            required = true,
+            defaultValue = "",
+            help = "specify the base path in ZooKeeper for agent configs")
+    private String zkBasePath;
+
+    public static FlumeConfig load(String yamlFile) throws IOException {
+        ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+        return mapper.readValue(new File(yamlFile), FlumeConfig.class);
+    }
+
+
+    public static FlumeConfig load(Map<String, Object> map) throws IOException {
+        ObjectMapper mapper = new ObjectMapper();
+        return mapper.readValue(new ObjectMapper().writeValueAsString(map), FlumeConfig.class);
+    }
+}

pulsar-io/flume/src/main/java/org/apache/pulsar/io/flume/FlumeConnector.java

@@ -0,0 +1,124 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.flume;
+
+import com.google.common.collect.Lists;
+import com.google.common.eventbus.EventBus;
+import org.apache.commons.cli.ParseException;
+import org.apache.flume.Constants;
+import org.apache.flume.lifecycle.LifecycleAware;
+import org.apache.flume.util.SSLUtil;
+import org.apache.pulsar.io.flume.node.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+
+public class FlumeConnector {
+
+    private static final Logger log = LoggerFactory
+            .getLogger(FlumeConnector.class);
+
+    protected Application application;
+
+    public void StartConnector(FlumeConfig flumeConfig) throws Exception {
+        SSLUtil.initGlobalSSLParameters();
+        String agentName = flumeConfig.getName();
+        boolean reload = !flumeConfig.getNoReloadConf();
+        boolean isZkConfigured = false;
+        if (flumeConfig.getZkConnString().length() > 0) {
+            isZkConfigured = true;
+        }
+        if (isZkConfigured) {
+            // get options
+            String zkConnectionStr = flumeConfig.getZkConnString();
+            String baseZkPath = flumeConfig.getZkBasePath();
+            if (reload) {
+                EventBus eventBus = new EventBus(agentName + "-event-bus");
+                List<LifecycleAware> components = Lists.newArrayList();
+                PollingZooKeeperConfigurationProvider zookeeperConfigurationProvider =
+                        new PollingZooKeeperConfigurationProvider(
+                                agentName, zkConnectionStr, baseZkPath, eventBus);
+                components.add(zookeeperConfigurationProvider);
+                application = new Application(components);
+                eventBus.register(application);
+            } else {
+                StaticZooKeeperConfigurationProvider zookeeperConfigurationProvider =
+                        new StaticZooKeeperConfigurationProvider(
+                                agentName, zkConnectionStr, baseZkPath);
+                application = new Application();
+                application.handleConfigurationEvent(zookeeperConfigurationProvider.getConfiguration());
+            }
+
+        } else {
+            File configurationFile = new File(flumeConfig.getConfFile());
+             /*
+         * The following is to ensure that by default the agent will fail on
+         * startup if the file does not exist.
+         */
+            if (!configurationFile.exists()) {
+                // If command line invocation, then need to fail fast
+                if (System.getProperty(Constants.SYSPROP_CALLED_FROM_SERVICE) ==
+                        null) {
+                    String path = configurationFile.getPath();
+                    try {
+                        path = configurationFile.getCanonicalPath();
+                    } catch (IOException ex) {
+                        log.error("Failed to read canonical path for file: " + path,
+                                ex);
+                    }
+                    throw new ParseException("The specified configuration file does not exist: " + path);
+                }
+            }
+            List<LifecycleAware> components = Lists.newArrayList();
+
+            if (reload) {
+                EventBus eventBus = new EventBus(agentName + "-event-bus");
+                PollingPropertiesFileConfigurationProvider configurationProvider =
+                        new PollingPropertiesFileConfigurationProvider(
+                                agentName, configurationFile, eventBus, 30);
+                components.add(configurationProvider);
+                application = new Application(components);
+                eventBus.register(application);
+            } else {
+                PropertiesFileConfigurationProvider configurationProvider =
+                        new PropertiesFileConfigurationProvider(agentName, configurationFile);
+                application = new Application();
+                application.handleConfigurationEvent(configurationProvider.getConfiguration());
+            }
+        }
+        application.start();
+
+        final Application appReference = application;
+        Runtime.getRuntime().addShutdownHook(new Thread("agent-shutdown-hook") {
+            @Override
+            public void run() {
+                appReference.stop();
+            }
+        });
+    }
+
+    public void stop() {
+        if (application != null) {
+            application.stop();
+        }
+    }
+}