[KAFKA-9546] Make FileStreamSourceTask extendable with generic streams

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Won't Fix
• Component/s: connect
• Labels: connect-api needs-kip
• Assignee: Csaba Galyo
• Reporter: Csaba Galyo
• Created: 13/Feb/20 11:09
• Updated: 29/Sep/20 15:48
• Resolved: 29/Sep/20 15:44

Description

Use case: I want to read a ZIP compressed text file with a file connector and send it to Kafka.

Currently, we have FileStreamSourceConnector which reads a \n delimited text file. This connector always returns a task of type FileStreamSourceTask.

The FileStreamSourceTask reads from stdio or opens a file InputStream. The issue with this approach is that the input needs to be a text file, otherwise it won't work. 

The code should be modified so that users could change the default InputStream to eg. ZipInputStream, or any other format. The code is currently written in such a way that it's not possible to extend it, we cannot use a different input stream. 

See example here where the code got copy-pasted just so it could read from a ZstdInputStream (which reads ZSTD compressed files): https://github.com/gcsaba2/kafka-zstd/tree/master/src/main/java/org/apache/kafka/connect/file

 

I suggest 2 changes:

1. FileStreamSourceConnector should be extendable to return tasks of different types. These types would be input by the user through the configuration map
2. FileStreamSourceTask should be modified so it could be extended and child classes could define different input streams.

Issue Links

• links to: GitHub Pull Request #8134