Repository: apache/kafka
File: connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStore.java (Java)

Code before:
            TopicAdmin topicAdmin
    ) {
        return new KafkaOffsetBackingStore(() -> topicAdmin, KafkaOffsetBackingStore::noClientId) {
            @Override
            public void configure(final WorkerConfig config) {
                this.exactlyOnce = config.exactlyOnceSourceEnabled();
                this.offsetLog = KafkaBasedLog.withExistingClients(
                        topic,
                        consumer,
                        null,
                        topicAdmin,
                        consumedCallback,
                        Time.SYSTEM,
                        initialize(topic, newTopicDescription(topic, config))
                );
            }
        };
    }

    private static String noClientId() {
        throw new UnsupportedOperationException("This offset store should not instantiate any Kafka clients");
    }

    protected KafkaBasedLog<byte[], byte[]> offsetLog;
    private final HashMap<ByteBuffer, ByteBuffer> data = new HashMap<>();
    private final Supplier<TopicAdmin> topicAdminSupplier;
    private final Supplier<String> clientIdBase;
    private SharedTopicAdmin ownTopicAdmin;
    protected boolean exactlyOnce;

    /**
     * Create an {@link OffsetBackingStore} backed by a Kafka topic. This constructor will cause the
     * store to instantiate and close its own {@link TopicAdmin} during {@link #configure(WorkerConfig)}
     * and {@link #stop()}, respectively.
     *
     * @deprecated use {@link #KafkaOffsetBackingStore(Supplier, Supplier)} instead
     */
    @Deprecated
    public KafkaOffsetBackingStore() {
        this.topicAdminSupplier = null;
        this.clientIdBase = () -> "connect-distributed-";
    }

    /**
     * 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

Review comment (suggestion):
```suggestion
    // Visible for testing
    final HashMap<ByteBuffer, ByteBuffer> data = new HashMap<>();
```

Code after:
            TopicAdmin topicAdmin
    ) {
        return new KafkaOffsetBackingStore(() -> topicAdmin, KafkaOffsetBackingStore::noClientId) {
            @Override
            public void configure(final WorkerConfig config) {
                this.exactlyOnce = config.exactlyOnceSourceEnabled();
                this.offsetLog = KafkaBasedLog.withExistingClients(
                        topic,
                        consumer,
                        null,
                        topicAdmin,
                        consumedCallback,
                        Time.SYSTEM,
                        initialize(topic, newTopicDescription(topic, config))
                );
            }
        };
    }

    private static String noClientId() {
        throw new UnsupportedOperationException("This offset store should not instantiate any Kafka clients");
    }

    protected KafkaBasedLog<byte[], byte[]> offsetLog;
    // Visible for testing
    final HashMap<ByteBuffer, ByteBuffer> data = new HashMap<>();
    private final Supplier<TopicAdmin> topicAdminSupplier;
    private final Supplier<String> clientIdBase;
    private SharedTopicAdmin ownTopicAdmin;
    protected boolean exactlyOnce;

    /**
     * Create an {@link OffsetBackingStore} backed by a Kafka topic. This constructor will cause the
     * store to instantiate and close its own {@link TopicAdmin} during {@link #configure(WorkerConfig)}
     * and {@link #stop()}, respectively.
     *
     * @deprecated use {@link #KafkaOffsetBackingStore(Supplier, Supplier)} instead
     */
    @Deprecated
    public KafkaOffsetBackingStore() {
        this.topicAdminSupplier = null;
        this.clientIdBase = () -> "connect-distributed-";
    }

    /**
     * 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