https://flink.apache.org/ logo
Join Slack
Powered by
Hello everyone! Need some help. Trying to upgrade...
# troubleshooting
о
Hello everyone! Need some help. Trying to upgrade flink job from flink 1.14.3 to 1.16.0. I have unexpected behaviour for FileSink. I have uint test which checks that job stored avro event into file. To trigger flushing data I’m triggering MiniClusterClient’s 
triggerSavepoint()
method, but I see that after savepoint created, the file in filesystem still is hidden and inprogress, like: 
build/junit15822280378836783130/archive-output/2_shards/0/2023/04/26/17/.part-0-0.inprogress.8a05ee6b-c969-4e15-88ba-9a3b78f46035
, but I’m expecting that file should be a permanent (closed, not hidden). I’ve did some attempts to find whats going on and how to fix it and one interesting thing I found: if increase timeout in test and job will trigger checkpoint storing (not savepoint triggered by hands, but checkpoint schedule in 1min) - so after it checkpoint everything as expected: file renamed and stored. 
23:41:38.838 [Checkpoint Timer] INFO  o.a.f.r.c.CheckpointCoordinator - Triggering checkpoint 1 (type=SavepointType{name='Savepoint', postCheckpointAction=NONE, formatType=CANONICAL}) @ 1682973698824 for job e63ff74fc4bc34a54c3f1f05855d149c.
23:41:38.948 [jobmanager-io-thread-3] INFO  o.a.f.r.c.CheckpointCoordinator - Completed checkpoint 1 for job e63ff74fc4bc34a54c3f1f05855d149c (15345 bytes, checkpointDuration=121 ms, finalizationTime=3 ms).
23:42:38.955 [Checkpoint Timer] INFO  o.a.f.r.c.CheckpointCoordinator - Triggering checkpoint 2 (type=CheckpointType{name='Checkpoint', sharingFilesStrategy=FORWARD_BACKWARD}) @ 1682973758954 for job e63ff74fc4bc34a54c3f1f05855d149c.
23:42:38.991 [jobmanager-io-thread-5] INFO  o.a.f.r.c.CheckpointCoordinator - Completed checkpoint 2 for job e63ff74fc4bc34a54c3f1f05855d149c (15388 bytes, checkpointDuration=33 ms, finalizationTime=4 ms).
23:42:39.528 [Test worker] INFO  c.s.d.t.h.FlinkHarness$Companion - Cancelling the job e63ff74fc4bc34a54c3f1f05855d149c...
Does anyone know why it could be? What changed and how I should use it to fix my tests? I hope that in production it should work with enabled checkpoints, but what about savepoints? Thanks a lot for any help and suggestions
Should add, that in Job’s class using sink:
Copy code
FileSink
        .forBulkFormat(Path(path), AvroBulkWriterFactory(ArchiveRecord::class.java))
        .withBucketAssigner(ArchiveRecordBucketAssigner(numOfBuckets))
        .build()
with custom implementation for BulkWriter:
Copy code
import org.apache.avro.Schema
import org.apache.avro.file.DataFileWriter
import org.apache.avro.generic.GenericDatumWriter
import org.apache.avro.generic.GenericRecord
import org.apache.avro.io.DatumWriter
import org.apache.avro.reflect.ReflectData
import org.apache.avro.reflect.ReflectDatumWriter
import org.apache.avro.specific.SpecificDatumWriter
import org.apache.avro.specific.SpecificRecordBase
import org.apache.flink.api.common.serialization.BulkWriter
import org.apache.flink.core.fs.FSDataOutputStream

class AvroBulkWriterFactory<IN>(
    private val avroValueType: Class<IN>,
    private val userDefinedSchema: Schema? = null,
    private val codecType: CodecType = CodecType.SNAPPY
) : BulkWriter.Factory<IN> {

    override fun create(out: FSDataOutputStream): BulkWriter<IN> {
        val datumWriter: DatumWriter<IN>
        val schema: Schema
        when {
            SpecificRecordBase::class.java.isAssignableFrom(avroValueType) -> {
                datumWriter = SpecificDatumWriter(avroValueType)
                schema = (avroValueType.newInstance() as SpecificRecordBase).schema
            }

            GenericRecord::class.java.isAssignableFrom(avroValueType) -> {
                checkNotNull(userDefinedSchema) { "Schema must be set when using Generic Record" }
                datumWriter = GenericDatumWriter(userDefinedSchema)
                schema = userDefinedSchema
            }

            else -> {
                datumWriter = ReflectDatumWriter(avroValueType)
                schema = ReflectData.get().getSchema(avroValueType)
            }
        }
        val dataFileWriter: DataFileWriter<IN> = DataFileWriter(datumWriter)
        dataFileWriter.setCodec(getCodecFactory(codecType))
        dataFileWriter.create(userDefinedSchema ?: schema, out)
        return AvroBulkWriter(dataFileWriter)
    }

    private class AvroBulkWriter<IN>(private val avroWriter: DataFileWriter<IN>) : BulkWriter<IN> {

        override fun addElement(element: IN) {
            avroWriter.append(element)
        }

        override fun flush() {
            avroWriter.flush()
        }

        override fun finish() {
            avroWriter.flush()
//            avroWriter.close()
        }
    }
}
2 Views
Open in Slack
PreviousNext
Powered by