@Override
    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
        stickyPartitionCache.nextPartition(topic, cluster, prevPartition);
    }
If a batch completed for the current sticky partition, change the sticky partition. Alternately, if
no sticky partition has been determined, set one.
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/PartitionerSelector.java