protected NewTopic newTopicDescription(final String topic, final WorkerConfig config) {
        Map<String, Object> topicSettings = config instanceof DistributedConfig
                ? ((DistributedConfig) config).offsetStorageTopicSettings()
                : Collections.emptyMap();
        return TopicAdmin.defineTopic(topic)
                .config(topicSettings) // first so that we override user-supplied settings as needed
                .compacted()
                .partitions(config.getInt(DistributedConfig.OFFSET_STORAGE_PARTITIONS_CONFIG))
                .replicationFactor(config.getShort(DistributedConfig.OFFSET_STORAGE_REPLICATION_FACTOR_CONFIG))
                .build();
    }
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