I do upgrade from flink 1.11 to 1.15.1 writing .gz files on S3. Since connectors.fs is now obsolete, I have to change the sink I've used before: EventuallyConsistentBucketingSink:
https://github.com/sjwiesman/flink/blob/83a6400e2587b067d08a64bc7e10edd4b57e71b4/flink-connectors/flink-connector-filesystem/src/main/java/org/apache/flink/streaming/connectors/fs/consistent/EventuallyConsistentBucketingSink.java
to something new, so I use the FileSink:
return FileSink.forRowFormat(
new Path(outputBasePath), new Encoder<T>() {
@Override
public void encode(T record, OutputStream stream)
throws IOException {
GzipParameters params = new GzipParameters();
params.setCompressionLevel(Deflater.BEST_COMPRESSION);
GzipCompressorOutputStream out = new GzipCompressorOutputStream(stream, params);
OBJECT_MAPPER.writeValue(out, record);
out.finish();
}
})
.withBucketAssigner(new BasePathBucketAssigner<>())
.withRollingPolicy(OnCheckpointRollingPolicy.build())
.build();
where outputBasePath is some s3:// URI. I got this error:
Exception in thread "main" java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS
at org.apache.flink.runtime.fs.hdfs.HadoopRecoverableWriter.<init>(HadoopRecoverableWriter.java:60)
at org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.createRecoverableWriter(HadoopFileSystem.java:215)
at org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createBucketWriter(FileSink.java:475)
at org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer(FileSink.java:466)
at org.apache.flink.connector.file.sink.FileSink.getCommittableSerializer(FileSink.java:175)
at org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo.toString(CommittableMessageTypeInfo.java:120)
What I'm missing here?