Check and decompress gzip file stream before processing as Avro file

[tanbt]

S3 Source connector might failed to source compressed (gzip) avro files from S3 -> Kafka.
This change expects to decompress the input stream before processing it as Avro data stream.

[2025-12-18 10:43:18,773] ERROR [s3_source|task-0] Error trying to open inputStream: Not an Avro data file. (io.aiven.kafka.connect.common.source.input.AvroTransformer:157)
org.apache.avro.InvalidAvroMagicException: Not an Avro data file.
	at org.apache.avro.file.DataFileStream.initialize(DataFileStream.java:111)
	at org.apache.avro.file.DataFileStream.<init>(DataFileStream.java:90)
	at io.aiven.kafka.connect.common.source.input.AvroTransformer$1.inputOpened(AvroTransformer.java:59)
	at io.aiven.kafka.connect.common.source.input.Transformer$StreamSpliterator.tryAdvance(Transformer.java:155)
	at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.lambda$initPartialTraversalState$0(StreamSpliterators.java:297)
	at java.base/java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.fillBuffer(StreamSpliterators.java:206)
	at java.base/java.util.stream.StreamSpliterators$AbstractWrappingSpliterator.doAdvance(StreamSpliterators.java:161)
	at java.base/java.util.stream.StreamSpliterators$WrappingSpliterator.tryAdvance(StreamSpliterators.java:303)
	at java.base/java.util.Spliterators$1Adapter.hasNext(Spliterators.java:669)
	at io.aiven.kafka.connect.common.source.AbstractSourceRecordIterator.hasNext(AbstractSourceRecordIterator.java:200)
	at io.aiven.kafka.connect.s3.source.S3SourceTask$1.hasNext(S3SourceTask.java:83)
	at org.apache.commons.collections4.iterators.FilterIterator.setNextObject(FilterIterator.java:174)
	at org.apache.commons.collections4.iterators.FilterIterator.hasNext(FilterIterator.java:86)
	at io.aiven.kafka.connect.common.source.AbstractSourceTask.tryAdd(AbstractSourceTask.java:187)
	at io.aiven.kafka.connect.common.source.AbstractSourceTask$2.run(AbstractSourceTask.java:131)
	at java.base/java.lang.Thread.run(Thread.java:1474)

After the changes, I tried again but couldn't reproduce the issue, and the data is available in the destination Kafka.

[jclarysse]

are mark/reset generally not supported by the input stream, or what are the conditions potentially leading to this?

[tanbt]

I see, since BufferedInputStream always support mark/reset, no need to check for markSupported.

[jclarysse]

the connector is supposed to support other file compression types, e.g. snappy and zstd. wondering if we should handle all possible magic bytes here, or if conditionally instantiating the right type of input stream could be done in a much easier way ie. based on file.compression.type config, wdyt?

[tanbt]

Using file.compression.type is more explicit, I think this is a better approach than detecting magic bytes of all compression types.

[jclarysse]

the variable name decompressedInput sounds wrong at this point, indeed it might be an input that is not compressed at all, or one that will get identified as a compressedInput. since the purpose of this variable is to instantiate a BufferInputStream that supports mark/reset, why not call it just bufferedInput instead?

[tanbt]

Good point! Renamed now.

[jclarysse]

Thanks @tanbt for your contribution!
We'd probably need to refine a few things please, see my comments.