@Override
    protected void finalOffsetCommit(boolean failed) {
        // It should still be safe to commit offsets since any exception would have
        // simply resulted in not getting more records but all the existing records should be ok to flush
        // and commit offsets. Worst case, task.commit() will also throw an exception causing the offset
        // commit to fail.
        submittedRecords.awaitAllMessages(
                workerConfig.getLong(WorkerConfig.OFFSET_COMMIT_TIMEOUT_MS_CONFIG),
                TimeUnit.MILLISECONDS
        );
        updateCommittableOffsets();
        commitOffsets();
    }
{@link WorkerTask} that uses a {@link SourceTask} to ingest data into Kafka.
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSourceTask.java