Как использовать источник taildir в Flume для добавления только самых новых строк файла .txt?

Недавно я задал вопрос Apache Flume - отправлять только новое содержимое файла

Я перефразирую вопрос, чтобы узнать больше и предоставить больше пользы будущим пользователям Flume.

Установка: два сервера, один с файлом .txt, к которому регулярно добавляются строки.

Цель: использовать источник TAILDIR для Flume, чтобы добавить последнюю записанную строку в файл на другом сервере.

Проблема: Всякий раз, когда в исходный файл добавляется новая строка данных, текущая конфигурация добавляет все в файле на сервере 1 к файлу на сервере 2. Это приводит к дублированию строк в файле 2 и неправильному воссозданию файла с сервера 1.

Конфигурация на сервере 1:

    #configure the agent
    agent.sources=r1
    agent.channels=k1
    agent.sinks=c1

    #using memort channel to hold upto 1000 events
    agent.channels.k1.type=memory
    agent.channels.k1.capacity=1000
    agent.channels.k1.transactionCapacity=100

    #connect source, channel,sink
    agent.sources.r1.channels=k1
    agent.sinks.c1.channel=k1

    #define source
    agent.sources.r1.type=TAILDIR
    agent.sources.r1.channels=k1
    agent.sources.r1.filegroups=f1

    agent.sources.r1.filegroups.f1=/home/tail_test_dir/test.txt
    agent.sources.r1.maxBackoffSleep=1000

    #connect to another box using avro and send the data
    agent.sinks.c1.type=avro
    agent.sinks.c1.hostname=10.10.10.4
    agent.sinks.c1.port=4545

Конфигурация на сервере 2:

    #configure the agent
    agent.sources=r1
    agent.channels=k1
    agent.sinks=c1

    #using memory channel to hold up to 1000 events
    agent.channels.k1.type=memory
    agent.channels.k1.capacity=1000
    agent.channels.k1.transactionCapacity=100

    #connect source, channel, sink
    agent.sources.r1.channels=k1
    agent.sinks.c1.channel=k1

    #here source is listening at the specified port using AVRO for data
    agent.sources.r1.type=avro
    agent.sources.r1.bind=0.0.0.0
    agent.sources.r1.port=4545

    #use file_roll and write file at specified directory
    agent.sinks.c1.type=file_roll
    agent.sinks.c1.sink.directory=/home/Flume_dump