Map<TopicPartition, OffsetAndMetadata> currentOffsets() {
        return Collections.unmodifiableMap(currentOffsets);
    }
Poll for new messages with the given timeout. Should only be invoked by the worker thread.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java