@Override
    public void start() {
        log.info("Starting KafkaOffsetBackingStore");
        try {
            offsetLog.start();
        } catch (UnsupportedVersionException e) {
            String message;
            if (exactlyOnce) {
                message = "Enabling exactly-once support for source connectors requires a Kafka broker version that allows "
                        + "admin clients to read consumer offsets. Please either disable the worker's exactly-once "
                        + "support for source connectors, or upgrade to a newer Kafka broker version.";
            } else {
                message = "When " + ConsumerConfig.ISOLATION_LEVEL_CONFIG + "is set to "
                        + IsolationLevel.READ_COMMITTED
                        + ", a Kafka broker version that allows admin clients to read consumer offsets is required. "
                        + "Please either reconfigure the worker or connector, or upgrade to a newer Kafka broker version.";
            }
            throw new ConnectException(message, e);
        }
        log.info("Finished reading offsets topic and starting KafkaOffsetBackingStore");
    }
Create an {@link OffsetBackingStore} backed by a Kafka topic. This constructor will use the given
{@link Supplier} to acquire a {@link TopicAdmin} that will be used for interactions with the backing
Kafka topic. The caller is expected to manage the lifecycle of that object, including
{@link TopicAdmin#close(Duration) closing} it when it is no longer needed.
@param topicAdmin a {@link Supplier} for the {@link TopicAdmin} to use for this backing store;
                  may not be null, and may not return null
@param clientIdBase a {@link Supplier} that will be used to create a
{@link CommonClientConfigs#CLIENT_ID_DOC client ID} for Kafka clients instantiated by this store;
                    may not be null, and may not return null, but may throw {@link UnsupportedOperationException}
                    if this offset store should not create its own Kafka clients
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java