gaborgsomogyi commented on code in PR #24914:
URL: https://github.com/apache/flink/pull/24914#discussion_r1709562167
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactService.java:
##########
@@ -173,4 +203,36 @@ private static CompactingFileWriter.Type
getWriterType(FileCompactor fileCompact
+ fileCompactor.getClass());
}
}
+
+ /**
+ * Creates a raw {@link BulkWriter} that writes the part files into the
compacted output stream
+ * without any added bytes. This is required for compacting compressed
part files that use a
+ * codec that supports concat, like GZIP or BZIP2.
+ */
+ private static class RawBulkWriterFactory implements
BulkWriter.Factory<String> {
Review Comment:
Is it possible to find out the exact type here?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactService.java:
##########
@@ -173,4 +203,36 @@ private static CompactingFileWriter.Type
getWriterType(FileCompactor fileCompact
+ fileCompactor.getClass());
}
}
+
+ /**
+ * Creates a raw {@link BulkWriter} that writes the part files into the
compacted output stream
+ * without any added bytes. This is required for compacting compressed
part files that use a
+ * codec that supports concat, like GZIP or BZIP2.
+ */
+ private static class RawBulkWriterFactory implements
BulkWriter.Factory<String> {
+
+ @Override
+ public BulkWriter<String> create(FSDataOutputStream out) {
+ return new BulkWriter<String>() {
+ @Override
+ public void addElement(String element) throws IOException {
+ out.write(element.getBytes());
+ }
+
+ @Override
+ public void flush() throws IOException {
+ out.flush();
+ }
+
+ /**
+ * Finish without closing the stream, which will be closed by
{@link
+ * OutputStreamBasedPartFileWriter#closeForCommit()}.
+ */
+ @Override
+ public void finish() throws IOException {
+ out.flush();
Review Comment:
Maybe this should be `sync`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
