Instead of using .compressionOptions(SchemaBuilder.lz4()), manually provide the correct compression map using .compressionOptions(Map<String, Object>).
.compressionOptions(SchemaBuilder.lz4())
.compressionOptions(Map<String, Object>)