OrcBulkWriter(Vectorizer<T> vectorizer, Writer writer) {
        this.vectorizer = checkNotNull(vectorizer);
        this.writer = checkNotNull(writer);
        this.rowBatch = vectorizer.getSchema().createRowBatch();

        // Configure the vectorizer with the writer so that users can add
        // metadata on the fly through the Vectorizer#vectorize(...) method.
        this.vectorizer.setWriter(this.writer);
    }
@Test
    void testOrcBulkWriter(@TempDir File outDir) throws Exception {
        final Properties writerProps = new Properties();
        writerProps.setProperty("orc.compress", "LZ4");

        final OrcBulkWriterFactory<Record> writer =
                new OrcBulkWriterFactory<>(
                        new RecordVectorizer(schema), writerProps, new Configuration());

        StreamingFileSink<Record> sink =
                StreamingFileSink.forBulkFormat(new Path(outDir.toURI()), writer)
                        .withBucketAssigner(new UniqueBucketAssigner<>("test"))
                        .withBucketCheckInterval(10000)
                        .build();

        try (OneInputStreamOperatorTestHarness<Record, Object> testHarness =
                new OneInputStreamOperatorTestHarness<>(new StreamSink<>(sink), 1, 1, 0)) {

            testHarness.setup();
            testHarness.open();

            int time = 0;
            for (final Record record : input) {
                testHarness.processElement(record, ++time);
            }

            testHarness.snapshot(1, ++time);
            testHarness.notifyOfCompletedCheckpoint(1);

            OrcBulkWriterTestUtil.validate(outDir, input);
        }
    }