@Override
    public boolean requiresCleanupOfRecoverableState() {
        // we can't clean up any state prior to commit
        // see discussion: https://github.com/apache/flink/pull/15599#discussion_r623127365
        return false;
    }
@Test
    public void testRequiresCleanupOfRecoverableState() {
        assertFalse(writer.requiresCleanupOfRecoverableState());
    }