hi ！

Some of the files in the compressed file are a description of the file
I want to exclude these description files
How do I do it

I am trying to using ftps connector using flume to fetch secured data.

My property file
`# Naming the components on the current agent.
FtpAgent.sources = ftps1
FtpAgent.channels = FileChannel
FtpAgent.sinks = HDFS

Describing/Configuring the source
FtpAgent.sources.ftps1.type = org.keedio.flume.source.ftp.source.Source
FtpAgent.sources.ftps1.client.source = ftps
FtpAgent.sources.ftps1.name.server = 192.168.1.10
FtpAgent.sources.ftps1.user = xxx
FtpAgent.sources.ftps1.password = xxx
FtpAgent.sources.ftps1.port = 21

FtpAgent.sources.ftps1.security.enabled = true
FtpAgent.sources.ftps1.security.cipher = SSL
FtpAgent.sources.ftps1.security.certificate.enabled = true
FtpAgent.sources.ftps1.path.keystore = /usr/hdp/current/flume-server/keystore.jks
#FtpAgent.sources.ftps1.path.keystore=/etc/ssl/certs/ca-certificates.crt
FtpAgent.sources.ftps1.store.pass = xxx

FtpAgent.sources.ftps1.flushlines = false
FtpAgent.sources.ftps1.chunk.size = 1024
FtpAgent.sources.ftps1.run.discover.delay=5000
FtpAgent.sources.ftps1.working.directory = /home/volumata/Desktop/ftp_sample
#FtpAgent.sources.ftps1.filter.pattern = pig_.
#FtpAgent.sources.ftps1.filter.pattern = .+.csv
FtpAgent.sources.ftps1.search.recursive = false
#FtpAgent.sources.ftps1.folder = /home/volumata/Desktop/ftp_sample

Describing/Configuring the sink
FtpAgent.sinks.HDFS.type = hdfs
FtpAgent.sinks.HDFS.hdfs.path = hdfs://xxx:8000/topics/flume_ftp_source
FtpAgent.sinks.HDFS.hdfs.fileType = DataStream
FtpAgent.sinks.HDFS.hdfs.writeFormat = Text
FtpAgent.sinks.HDFS.hdfs.batchSize = 1000
FtpAgent.sinks.HDFS.hdfs.rollSize = 100
FtpAgent.sinks.HDFS.hdfs.rollCount = 100000
FtpAgent.sinks.hdfs.serializer=Text

Describing/Configuring the channel
FtpAgent.channels.FileChannel.type = file
FtpAgent.channels.FileChannel.capacity = 100000`

I am running the agent using this command,

bin/flume-ng agent --conf ./conf/ -f conf/ftps_flume_source.conf Dflume.root.logger=DEBUG,console -n FtpAgent

I am getting this error while running the agent

ERROR source.Source: Exception thrown in process, try to reconnect 0
javax.net.ssl.SSLException: Connection has been shutdown: javax.net.ssl.SSLException: Unsupported record version Unknown-48.48

I am not sure where i did mistakes,please anyone help me resolve this issue.

From the available configurations for the ftp source, I was not able to find a property via which I configure the path to the directory from which I want to pull files.

This would mean that I would need to host files I want to handle differently in separate users home directory.

I might be misreading the documentation but I guess this is an essential feature that is required.

Thanks.

I think we should set up a CI pipeline for this project using something like Travis. A label in the README that shows the status of the current build will ensure that users don't clone a failing build.
Thoughts?

on retrieving stream from soft link , inputstream points nowhere

I am trying to connect flume with ftp source, i had referred this community https://github.com/keedio/flume-ftp-source.

I have downloaded the ftp source jar for flume in this site and followed those instructions which is mentioned. Tried to connect with FTP.

Here it is pointing to all the files which is present in the home directory, INFO source.Source: Actual dir: /home/xxx files: 31.

I want to point only one file which is present in desktop, for that i have given working.directory property. But it is fetching all the records which is present in home directory.

And inbetween it is throwing some errors while pushing data to hdfs.

18/05/23 15:56:57 ERROR source.Source: ChannelException org.apache.flume.ChannelException: Unable to put event on required channel: org.apache.flume.channel.MemoryChannel{name: MemChannel} at org.apache.flume.channel.ChannelProcessor.processEvent(ChannelProcessor.java:275) at org.keedio.flume.source.ftp.source.Source.processMessage(Source.java:404) at org.keedio.flume.source.ftp.source.Source.readStream(Source.java:376) at org.keedio.flume.source.ftp.source.Source.discoverElements(Source.java:248) at org.keedio.flume.source.ftp.source.Source.discoverElements(Source.java:196) at org.keedio.flume.source.ftp.source.Source.discoverElements(Source.java:196) at org.keedio.flume.source.ftp.source.Source.discoverElements(Source.java:196) at org.keedio.flume.source.ftp.source.Source.discoverElements(Source.java:196) at org.keedio.flume.source.ftp.source.Source.discoverElements(Source.java:196) at org.keedio.flume.source.ftp.source.Source.process(Source.java:102) at org.apache.flume.source.PollableSourceRunner$PollingRunner.run(PollableSourceRunner.java:137) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.flume.ChannelException: Cannot commit transaction. Heap space limit of 55297reached. Please increase heap space allocated to the channel as the sinks may not be keeping up with the sources at org.apache.flume.channel.MemoryChannel$MemoryTransaction.doCommit(MemoryChannel.java:126) at org.apache.flume.channel.BasicTransactionSemantics.commit(BasicTransactionSemantics.java:151) at org.apache.flume.channel.ChannelProcessor.processEvent(ChannelProcessor.java:267) ... 11 more 18/05/23 15:56:57 INFO hdfs.BucketWriter: Creating xxx/topics/Specified_flume_ftp_data/FlumeData.1527071214656.tmp 18/05/23 15:56:57 INFO hdfs.BucketWriter: Closing xxx/topics/Specified_flume_ftp_data/FlumeData.1527071214656.tmp 18/05/23 15:56:57 INFO hdfs.BucketWriter: Renaming xxx/topics/Specified_flume_ftp_data/FlumeData.1527071214656.tmp to xxx/topics/Specified_flume_ftp_data/FlumeData.1527071214656

What property need to be added to point only particular directory?

My configuration property file

`# Naming the components on the current agent.
FtpAgent.sources = ftp1
FtpAgent.channels = MemChannel
FtpAgent.sinks = HDFS

Describing/Configuring the source

FtpAgent.sources.ftp1.type = org.keedio.flume.source.ftp.source.Source
FtpAgent.sources.ftp1.client.source = ftp
FtpAgent.sources.ftp1.name.server = 192.168.1.1
FtpAgent.sources.ftp1.user = xxxx
FtpAgent.sources.ftp1.password = xxxx
FtpAgent.sources.ftp1.port = 21
FtpAgent.sources.ftp1.flushlines = false
FtpAgent.sources.ftp1.chunk.size = 1024
FtpAgent.sources.fpt1.run.discover.delay=5000
FtpAgent.sources.fpt1.working.directory = /home/xxx/Desktop/ftp_sample

Describing/Configuring the sink

FtpAgent.sinks.HDFS.type = hdfs
FtpAgent.sinks.HDFS.hdfs.path = hdfs://xxx/topics/Specified_flume_ftp_data
FtpAgent.sinks.HDFS.hdfs.fileType = DataStream
FtpAgent.sinks.HDFS.hdfs.writeFormat = Text
FtpAgent.sinks.HDFS.hdfs.batchSize = 1000
FtpAgent.sinks.HDFS.hdfs.rollSize = 100
FtpAgent.sinks.HDFS.hdfs.rollCount = 100000
FtpAgent.sinks.hdfs.serializer=Text

Describing/Configuring the channel

FtpAgent.channels.MemChannel.type = memory
FtpAgent.channels.MemChannel.capacity = 100000
FtpAgent.channels.MemChannel.transactionCapacity = 1000
FtpAgent.channels.MemChannel.byteCapacity = 6912212

Binding the source and sink to the channel

FtpAgent.sources.ftp1.channels = MemChannel
FtpAgent.sinks.HDFS.channel = MemChannel

I am running the agent by hitting this below command,

bin/flume-ng agent --conf ./conf/ -f conf/flume_ftp_source.conf Dflume.root.logger=DEBUG,console -n FtpAgent
`

Hi,

I require the name of the file being streamed from the FTPSource in my custom interceptor.
Is it a good idea to add the file name being processed in the event header in FTPSource so that we get that in the custom interceptor.

Thanks
Deepak

Hi,

I am currently using the Flume FTP source to consume events from an FTP location. Since, the current FTP Flume source reads and processes one file at a time in a sequential manner it is unable to catch up with the speed at which events are arriving at the FTP location. Is there a way to speed up this by creating parallel FTP sources to consume from the same FTP location and speed this up.

Thanks
Deepak

In Jenkins, when running EmbeddedFTPSourceTest.testFtpFailure, simulating the FTP server failure leads to an infinite number of connection retries, since the call to

ftpSourceUtils.connectToserver() 

(in https://github.com/keedio/flume-ftp-source/blob/1.0.1/src/main/java/org/apache/flume/source/FTPSource.java#L103), throws an exception and the counter is never incremented.

Hi,

The text below is more a solution of an issue than an issue without solution.

Because I needed to store FTP files to HDFS by keeping the source file name for every FTP file read, I found unfortunately that the value of the property %{basename} was empty !

What a pity ! Indeed sometimes the name of the source file can contains very useful informations you can extract and inject into the target file (into its path or its content), like transaction dates, business codifications, and even more...

So first I developed a Flume Interceptor thinking that it be would nicer, but nothing has changed !
The %{basename} property was empty yet !

Finally, I adapted your Java class named [Source] in order to propagate the property related on the source file name (%{basename}), and this for every FTP file read.
So know it works well !

For information:
I added the parameter 'elementName' to the methods 'readStream(..)' and 'processMessages(..)'.

I added the following line into the 'processMessages(..)' class:
headers.put("basename", elementName);

Finally, I added the parameter 'elementName' by modifying the following code into the 'discoverElements(..)' method:
if (!readStream(inputStream, position, elementName)) {
inputStream = null;
}

Please see below the content of the new class [Source] to replace by the current class if someone need to use the %{basename} property (the source file name for every FTP source file read):

/*
 * KEEDIO
 */
package org.keedio.flume.source.ftp.source;

import java.util.*;`

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.PollableSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.flume.ChannelException;

import java.io.IOException;
import java.io.InputStream;
import java.io.ByteArrayOutputStream;

import org.keedio.flume.source.ftp.source.utils.FTPSourceEventListener;

import org.keedio.flume.source.ftp.metrics.SourceCounter;
import java.util.List;

import java.io.BufferedReader;
import java.io.InputStreamReader;

import org.keedio.flume.source.ftp.client.factory.SourceFactory;
import org.keedio.flume.source.ftp.client.KeedioSource;
import java.nio.charset.Charset;
import org.apache.flume.source.AbstractSource;

/**
 *
 * @author luislazaro [email protected] - KEEDIO
 *
 */
public class Source extends AbstractSource implements Configurable, PollableSource {

    private SourceFactory sourceFactory = new SourceFactory();
    private KeedioSource keedioSource;

    private static final Logger LOGGER = LoggerFactory.getLogger(Source.class);
    private static final short ATTEMPTS_MAX = 3; //  max limit attempts reconnection
    private static final long EXTRA_DELAY = 10000;
    private int counterConnect = 0;
    private FTPSourceEventListener listener = new FTPSourceEventListener();
    private SourceCounter sourceCounter;

    
    /**
     * Request keedioSource to the factory
     *
     * @param context
     * @return KeedioSource
     */
    public KeedioSource orderKeedioSource(Context context) {
        keedioSource = sourceFactory.createKeedioSource(context);
        return keedioSource;
    }

    /**
     *
     * @param context
     */
    @Override
    public void configure(Context context) {
        keedioSource = orderKeedioSource(context);
        if (keedioSource.existFolder()) {
            keedioSource.makeLocationFile();
        } else {
            LOGGER.error("Folder " + keedioSource.getPathTohasmap().toString() + " not exists");
        }
        keedioSource.connect();
        sourceCounter = new SourceCounter("SOURCE." + getName());
        keedioSource.checkPreviousMap();
    }

    /**
     * @return Status , process source configured from context
     * @throws org.apache.flume.EventDeliveryException
     */
    @Override
    public PollableSource.Status process() throws EventDeliveryException {
        
        try {            
            LOGGER.info("Actual dir:  " + keedioSource.getDirectoryserver() + " files: "
                    + keedioSource.getFileList().size());
            
            discoverElements(keedioSource, keedioSource.getDirectoryserver(), "", 0);
            keedioSource.cleanList(); //clean list according existing actual files
            keedioSource.getExistFileList().clear();
        } catch (IOException e) {
            LOGGER.error("Exception thrown in proccess, try to reconnect " + counterConnect, e);

            if (!keedioSource.connect()) {
                counterConnect++;
            } else {
                keedioSource.checkPreviousMap();
            }

            if (counterConnect < ATTEMPTS_MAX) {
                process();
            } else {
                LOGGER.error("Server connection closed without indication, reached limit reconnections " + counterConnect);
                try {
                    Thread.sleep(keedioSource.getRunDiscoverDelay() + EXTRA_DELAY);
                    counterConnect = 0;
                } catch (InterruptedException ce) {
                    LOGGER.error("InterruptedException", ce);
                }
            }
        }
        keedioSource.saveMap();

        try {
            Thread.sleep(keedioSource.getRunDiscoverDelay());
            return PollableSource.Status.READY;     //source was successfully able to generate events
        } catch (InterruptedException inte) {
            LOGGER.error("Exception thrown in process while putting to sleep", inte);
            return PollableSource.Status.BACKOFF;   //inform the runner thread to back off for a bit
        }
    }

    /**
     * @return void
     */
    @Override
    public synchronized void start() { 
        LOGGER.info("Starting Keedio source ...", this.getName());
        LOGGER.info("Source {} starting. Metrics: {}", getName(), sourceCounter);
        super.start();
        sourceCounter.start();
    }

    /**
     * @return void
     */
    @Override
    public synchronized void stop() {
        keedioSource.saveMap();
        if (keedioSource.isConnected()) {
            keedioSource.disconnect();
        }
        sourceCounter.stop();
        super.stop();
    }

    /**
     * discoverElements: find files to process them
     *
     * @param <T>
     * @param keedioSource
     * @param parentDir, will be the directory retrieved by the server when
     * connected
     * @param currentDir, actual dir in the recursive method
     * @param level, deep to search
     * @throws IOException
     */
    // @SuppressWarnings("UnnecessaryContinue")
    public <T> void discoverElements(KeedioSource keedioSource, String parentDir, String currentDir, int level) throws IOException {

        long position = 0L;

        String dirToList = parentDir;
        if (!("").equals(currentDir)){
            dirToList += "/" + currentDir;
        } 
        List<T> list = keedioSource.listElements(dirToList);
        if (!(list.isEmpty())) {

            for (T element : list) {
                String elementName = keedioSource.getObjectName(element);
                if (elementName.equals(".") || elementName.equals("..")) {
                    continue;
                }

                if (keedioSource.isDirectory(element)) {
                    LOGGER.info("[" + elementName + "]");
                    keedioSource.changeToDirectory(parentDir);
                    discoverElements(keedioSource, dirToList, elementName, level + 1);

                } else if (keedioSource.isFile(element)) { //element is a regular file
                    keedioSource.changeToDirectory(dirToList);
                    keedioSource.getExistFileList().add(dirToList + "/" + elementName);  //control of deleted files in server  

                    //test if file is new in collection
                    if (!(keedioSource.getFileList().containsKey(dirToList + "/" + elementName))) { //new file
                        sourceCounter.incrementFilesCount(); //include all files, even not yet processed
                        position = 0L;
                        LOGGER.info("Discovered: " + elementName + " ,size: " + keedioSource.getObjectSize(element));
                    } else { //known file
                        long prevSize = (long) keedioSource.getFileList().get(dirToList + "/" + elementName);
                        position = prevSize;
                        long dif = keedioSource.getObjectSize(element) - (long) keedioSource.getFileList().get(dirToList + "/" + elementName);

                        if (dif > 0) {
                            LOGGER.info("Modified: " + elementName + " ,size: " + dif);
                        } else if (dif < 0) { //known and full modified
                            keedioSource.getExistFileList().remove(dirToList + "/" + elementName); //will be rediscovered as new file
                            keedioSource.saveMap();
                            continue;
                        } else {
                            continue;
                        }

                    } //end if known file

                    //common for all regular files
                    InputStream inputStream = null;
                    try {
                        inputStream = keedioSource.getInputStream(element);
                        listener.fileStreamRetrieved();

                        if (!readStream(inputStream, position, elementName)) {
                            inputStream = null;
                        }

                        boolean success = inputStream != null && keedioSource.particularCommand(); //mandatory if FTPClient
                        if (success) {
                            keedioSource.getFileList().put(dirToList + "/" + elementName, keedioSource.getObjectSize(element));
                            keedioSource.saveMap();

                            if (position != 0) {
                                sourceCounter.incrementCountModProc();
                            } else {
                                sourceCounter.incrementFilesProcCount();
                            }

                            LOGGER.info("Processed:  " + elementName + " ,total files: " + this.keedioSource.getFileList().size() + "\n");

                        } else {
                            handleProcessError(elementName);
                        }
                    } catch (IOException e) {
                        handleProcessError(elementName);
                        LOGGER.error("Failed retrieving inputStream on discoverElements ", e);
                        continue;
                    }

                    keedioSource.changeToDirectory(dirToList);

                } else if (keedioSource.isLink(element)) {
                    LOGGER.info(elementName + " is a link of " + this.keedioSource.getLink(element) + " could not retrieve size");
                    keedioSource.changeToDirectory(parentDir);
                    continue;
                } else {
                    LOGGER.info(elementName + " unknown type of file");
                    keedioSource.changeToDirectory(parentDir);
                    continue;
                }
                keedioSource.changeToDirectory(parentDir);

            } 
        } 
    }

    /**
     * Read retrieved stream from ftpclient into byte[] and process. If
     * flushlines is true the retrieved inputstream will be readed by lines. And
     * the type of file is set to ASCII from KeedioSource.
     *
     * @return boolean
     * @param inputStream
     * @param position
     */
    public boolean readStream(InputStream inputStream, long position, String elementName) {
        if (inputStream == null) {
            return false;
        }

        boolean successRead = true;

        if (keedioSource.isFlushLines()) {
            try {
                inputStream.skip(position);
                try (BufferedReader in = new BufferedReader(new InputStreamReader(inputStream, Charset.defaultCharset()))) {
                    String line = null;

                    while ((line = in.readLine()) != null) {
                        processMessage(line.getBytes(), elementName);
                    }

                }
                inputStream.close();
            } catch (IOException e) {
                LOGGER.error(e.getMessage(), e);
                successRead = false;
            }
        } else {

            try {
                inputStream.skip(position);
                int chunkSize = keedioSource.getChunkSize();
                byte[] bytesArray = new byte[chunkSize];
                int bytesRead = -1;
                while ((bytesRead = inputStream.read(bytesArray)) != -1) {
                    try (ByteArrayOutputStream baostream = new ByteArrayOutputStream(chunkSize)) {
                        baostream.write(bytesArray, 0, bytesRead);
                        byte[] data = baostream.toByteArray();
                        processMessage(data, elementName);
                    }
                }

                inputStream.close();
            } catch (IOException e) {
                LOGGER.error("on readStream", e);
                successRead = false;

            }
        }
        return successRead;
    }

    /**
     * @void process last appended data to files
     * @param lastInfo byte[]
     */
    public void processMessage(byte[] lastInfo, String elementName) {
        byte[] message = lastInfo;
        Event event = new SimpleEvent();
        Map<String, String> headers = new HashMap<>();
        headers.put("timestamp", String.valueOf(System.currentTimeMillis()));
        headers.put("basename", elementName);
        event.setBody(message);
        event.setHeaders(headers);
        try {
            getChannelProcessor().processEvent(event);
        } catch (ChannelException e) {
            LOGGER.error("ChannelException", e);
        }
        sourceCounter.incrementCountSizeProc(message.length);
        sourceCounter.incrementEventCount();
    }

    /**
     * @param listener
     */
    public void setListener(FTPSourceEventListener listener) {
        this.listener = listener;
    }

    /**
     * @param fileName
     */
    public void handleProcessError(String fileName) {
        LOGGER.info("failed retrieving stream from file, will try in next poll :" + fileName);
        sourceCounter.incrementFilesProcCountError();
    }

    /**
     *
     * @param ftpSourceCounter
     */
    public void setFtpSourceCounter(SourceCounter ftpSourceCounter) {
        this.sourceCounter = ftpSourceCounter;
    }

    /**
     * @return KeedioSource
     */
    public KeedioSource getKeedioSource() {
        return keedioSource;
    }

} //endclass

I want to know the files name in event process, is there any way to put the file's name into event headers?

Despite the tcp connection is in state ESTABLISHED and Recv-Q is active, the inputStream may become indefinitely retrieved, because the end of line seems no to be reached.
This bug could be related with official oracle-bug:
http://bugs.java.com/bugdatabase/view_bug.do?bug_id=4262765

Hi, I am trying to poll an FTP location for a zip file. Since i need the to unzip the zip file and send the XML file, i wrote an interceptor to do the task. Although interceptor works, the zip file does not get through properly from the FTPSource to Intercpetor. If i output the file in FTPSource.readstream at the beginning before converting the inputstream to byte array, the file gets correctly created. But after converting to byte array if i output the file, it does not get created properly.
The scenario i am working on is, collect the zip file in Flume , unzip the file, extract the XML file and send it to Kafka for further processing. Can you please suggest the best way of doing this.

Thanks
Deepak

Actually memory is -Xms20m for HashMap.

Hi,
I am trying to use the Flume FTP source which is using the Apache Flume 1.5.2 installed using Ambari. When the flume FTPSource is executed, after reading the file from the FTP source i get the below error
Error saving map File java.io.FileNotFoundException: hasmap.ser (Permission denied)

I presume the file is written under the /home/ directory
I have checked the permissions and it seems to be fine
Can you please help me solve this.

Thanks
Deepak

Hi,

our client ftp server as has several sets of files. I am trying to select a particular source files. It is ftping all the files irrespective of file name or regular expression i have provided? there there property i can set to get a particular file set? I am wondering if there are ways to get file along with file name?

I am using fallowing flume config file property to filer file name.
Agent.sources.sftp1.file.name = WeatherForecast10092015.csv
or
Agent.sources.sftp1.file.name = WeatherForecast*

Any help is appreciated.

Thanks
Chandra

It looks like the directory is not changed if the working.directory is set when using the SFTPSource. I have a PR I'll submit for this.

after visiting last directory in pool, prcess hooks int that directory

Hi,

I require the name of the file being streamed from the FTPSource in my custom interceptor.
Is it a good idea to add the file name being processed in the event header in FTPSource so that we get that in the custom interceptor.

Thanks
Deepak

Generally,
read(byte[] input, int offset, int length) returns –1 on end of stream; but if
length is 0, then it does not notice the end of stream and returns 0 instead

HI

I am trying to use flume flume first time. I need to connect to ftp server to transfer files using port 22. I would like to take advantage of your plug in. it is thoroughgoing errors while i am trying to build my it. it is complaining about missing files. Could you please support me?

C:\github\flume-ftp-source-master\flume-ftp-source-master\src\main\java\org\keedio\flume\source\ftp\client\sources\FTPSource.java
Error:(26, 44) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(35, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(101, 44) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftp\EmbeddedFtpSourceTest.java
Error:(27, 2) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(59, 29) java: multi-catch statement is not supported in -source 1.3
(use -source 7 or higher to enable multi-catch statement)
Error:(122, 32) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(204, 43) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
Error:(255, 31) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\client\KeedioSourceTest.java
Error:(48, 12) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(48, 45) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
Error:(66, 41) java: multi-catch statement is not supported in -source 1.3
(use -source 7 or higher to enable multi-catch statement)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\main\java\org\keedio\flume\source\ftp\metrics\SourceCounter.java
Error:(54, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\TestFileUtilsTest.java
Error:(20, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(65, 17) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(99, 25) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\main\java\org\keedio\flume\source\ftp\source\Source.java
Error:(68, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(163, 12) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(174, 28) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
Error:(279, 21) java: try-with-resources is not supported in -source 1.3
(use -source 7 or higher to enable try-with-resources)
Error:(324, 51) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftp\ExternalFtpSourceTests.java
Error:(17, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftps\server\EmbeddedFTPSserver.java
Error:(65, 30) java: multi-catch statement is not supported in -source 1.3
(use -source 7 or higher to enable multi-catch statement)
Error:(70, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\TestFileUtils.java
Error:(63, 23) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(68, 53) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
Error:(132, 22) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\main\java\org\keedio\flume\source\ftp\client\KeedioSource.java
Error:(38, 35) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(42, 54) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
Error:(221, 17) java: try-with-resources is not supported in -source 1.3
(use -source 7 or higher to enable try-with-resources)
Error:(272, 30) java: multi-catch statement is not supported in -source 1.3
(use -source 7 or higher to enable multi-catch statement)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftp\server\EmbeddedFTPServer.java
Error:(52, 29) java: multi-catch statement is not supported in -source 1.3
(use -source 7 or higher to enable multi-catch statement)
Error:(57, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ssh\AbstractSshSourceTest.java
Error:(12, 8) java: static import declarations are not supported in -source 1.3
(use -source 5 or higher to enable static import declarations)
Error:(31, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(107, 29) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(108, 21) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\main\java\org\keedio\flume\source\ftp\client\sources\SFTPSource.java
Error:(25, 45) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(54, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(182, 56) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\sshd\server\EmbeddedSSHDServer.java
Error:(56, 48) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(72, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\main\java\org\keedio\flume\source\ftp\client\sources\FTPSSource.java
Error:(28, 45) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(53, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(118, 44) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftp\AbstractFtpSourceTest.java
Error:(11, 8) java: static import declarations are not supported in -source 1.3
(use -source 5 or higher to enable static import declarations)
Error:(26, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(101, 29) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(102, 21) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftps\AbstractFtpsSourceTest.java
Error:(11, 8) java: static import declarations are not supported in -source 1.3
(use -source 5 or higher to enable static import declarations)
Error:(27, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(109, 29) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(110, 21) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\ftps\EmbeddedFtpsSourceTest.java
Error:(31, 2) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(63, 29) java: multi-catch statement is not supported in -source 1.3
(use -source 7 or higher to enable multi-catch statement)
Error:(126, 32) java: generics are not supported in -source 1.3
(use -source 5 or higher to enable generics)
Error:(207, 43) java: diamond operator is not supported in -source 1.3
(use -source 7 or higher to enable diamond operator)
Error:(258, 31) java: enhanced for loops are not supported in -source 1.3
(use -source 5 or higher to enable for-each loops)
C:\github\flume-ftp-source-master\flume-ftp-source-master\src\test\java\org\keedio\flume\source\ftp\source\SourceTest.java
Error:(20, 8) java: static import declarations are not supported in -source 1.3
(use -source 5 or higher to enable static import declarations)
Error:(40, 6) java: annotations are not supported in -source 1.3
(use -source 5 or higher to enable annotations)
Error:(252, 21) java: try-with-resources is not supported in -source 1.3
(use -source 7 or higher to enable try-with-resources)

Hello,

it seems there is a repeated typo in the README file :
agent.sources.ftp1.client.sources = ftp
...
agent.sources.ftps1.client.sources = ftps
...
agent.sources.sftp1.client.sources = sftp

=> there is an extra unwanted 's' ...client.sources.

The good news is that the example conf files are correct :)

Now only saves state on stoping main process:

commented for increasing speed discovering new elements