I'm attempting to use the Kafka file pulse connector (https://github.com/streamthoughts/kafka-connect-file-pulse) to read data from a file into a Kafka topic.
I start the connector using :
../bin/connect-standalone filepulse.properties connect-standalone.properties
Content of filepulse.properties:
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storage.file.filename=/opt/kafka-tmp/connect.offsets
bootstrap.servers=localhost:9092
plugin.path=/Users/user1234/connectors/
Content of connect-standalone.properties:
offset.storage.file.filename=/tmp/connect.offsets
# Flush much faster than normal, which is useful for testing/debugging
offset.flush.interval.ms=10000
name=CsvSchemaSpoolDir
halt.on.error=false
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storage.file.filename=/opt/kafka-tmp/connect.offsets
plugin.path=/Users/plugins/connectors/
connector.class= io.streamthoughts.kafka.connect.filepulse.source.FilePulseSourceConnector
fs.cleanup.policy.class= io.streamthoughts.kafka.connect.filepulse.clean.LogCleanupPolicy
fs.scanner.class= io.streamthoughts.kafka.connect.filepulse.scanner.local.LocalFSDirectoryWalker
fs.scan.directory.path=/opt/kafka-tmp/dir-to-process
fs.scan.filters= io.streamthoughts.kafka.connect.filepulse.scanner.local.filter.RegexFileListFilter
file.filter.regex.pattern=test.csv
fs.scan.interval.ms= 3000
internal.kafka.reporter.bootstrap.servers=localhost:9092
internal.kafka.reporter.id= connect-file-pulse-log4j-quickstart
internal.kafka.reporter.topic= connect-file-pulse-status
offset.strategy=name
read.max.wait.ms=5000
task.reader.class=io.streamthoughts.kafka.connect.filepulse.reader.RowFileInputReader
topic= logs-kafka-connect
tasks.max= 1
When I add data to test.csv the data is not sent to the topic logs-kafka-connect as per the configuration.
Startup of ../bin/connect-standalone filepulse.properties connect-standalone.properties looks fine as the file test.csv is detected :
[2021-01-01 21:01:26,800] INFO Waiting 2821 ms to scan for new files. (io.streamthoughts.kafka.connect.filepulse.source.FileSystemMonitorThread:87)
[2021-01-01 21:01:29,625] INFO Scanning local file system directory '/opt/kafka-tmp/dir-to-process' (io.streamthoughts.kafka.connect.filepulse.scanner.LocalFileSystemScanner:222)
[2021-01-01 21:01:29,629] INFO Completed scanned, number of files detected '1' (io.streamthoughts.kafka.connect.filepulse.scanner.LocalFileSystemScanner:224)
[2021-01-01 21:01:29,815] INFO Finished lookup for new files : '0' files selected (io.streamthoughts.kafka.connect.filepulse.scanner.LocalFileSystemScanner:229)
[2021-01-01 21:01:29,815] INFO Waiting 2810 ms to scan for new files. (io.streamthoughts.kafka.connect.filepulse.source.FileSystemMonitorThread:87)
I have explicitly tested producing and subscribing messages to the topic logs-kafka-connect not using the plugin and it behaves as expected, messages are added and can be read from the topic. Therefore, it appears I’ve not configured the plugin correctly.
The topic I’ve configured with the plugin : logs-kafka-connect is created by the plugin but adding messages to the file it's configured to listen on (test.csv) is not sending the messages to the topic. How to configure the plugin so that items that are added to test.csv are sent to the topic logs-kafka-connect ?
Update:
It does seem to me now that Kafka is not intended for this use case which is streaming updates of files to Kafka topics. I will use filebeat to accomplish my goal, consolidation of multiple log files to enable easier log file inspection.