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

Code before:
    public void setUp() throws Exception {
        Supplier<SharedTopicAdmin> adminSupplier = () -> {
            fail("Should not attempt to instantiate admin in these tests");
            // Should never be reached; only add this thrown exception to satisfy the compiler
            throw new AssertionError();
        };
        Supplier<String> clientIdBase = () -> CLIENT_ID_BASE;
        store = PowerMock.createPartialMock(
                KafkaOffsetBackingStore.class,
                new String[] {"createKafkaBasedLog"},
                adminSupplier, clientIdBase
        );
    }

    @Test
    public void testStartStop() throws Exception {
        expectConfigure();
        expectStart(Collections.emptyList());
        expectStop();
        expectClusterId();

        PowerMock.replayAll();

        Map<String, String> settings = new HashMap<>(DEFAULT_PROPS);
        settings.put("offset.storage.min.insync.replicas", "3");
        settings.put("offset.storage.max.message.bytes", "1001");
        store.configure(new DistributedConfig(settings));
        assertEquals(TOPIC, capturedTopic.getValue());
        assertEquals("org.apache.kafka.common.serialization.ByteArraySerializer", capturedProducerProps.getValue().get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG));
        assertEquals("org.apache.kafka.common.serialization.ByteArraySerializer", capturedProducerProps.getValue().get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG));
        assertEquals("org.apache.kafka.common.serialization.ByteArrayDeserializer", capturedConsumerProps.getValue().get(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG));
        assertEquals("org.apache.kafka.common.serialization.ByteArrayDeserializer", capturedConsumerProps.getValue().get(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG));

        assertEquals(TOPIC, capturedNewTopic.getValue().name());
        assertEquals(TOPIC_PARTITIONS, capturedNewTopic.getValue().numPartitions());
        assertEquals(TOPIC_REPLICATION_FACTOR, capturedNewTopic.getValue().replicationFactor());

        store.start();
        store.stop();

        PowerMock.verifyAll();
    }

    @Test
    public void testReloadOnStart() throws Exception {
        expectConfigure();
        expectStart(Arrays.asList(
                new ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0, 0, TP0_KEY.array(), TP0_VALUE.array(),
                    new RecordHeaders(), Optional.empty()),
                new ConsumerRecord<>(TOPIC, 1, 0, 0L, TimestampType.CREATE_TIME, 0, 0, TP1_KEY.array(), TP1_VALUE.array(),
                    new RecordHeaders(), Optional.empty()),

Review comment (suggestion):
We should also make sure that none of the tests resulted in unexpected calls to start, stop, etc. the underlying `KafkaBasedLog`:
```suggestion
        connectUtilsMockedStatic.close();
        Mockito.verifyNoMoreInteractions(storeLog);
```

Code after:
    @Captor
    private ArgumentCaptor<Map<String, Object>> capturedProducerProps;
    @Captor
    private ArgumentCaptor<Map<String, Object>> capturedConsumerProps;
    @Captor
    private ArgumentCaptor<Supplier<TopicAdmin>> capturedAdminSupplier;
    @Captor
    private ArgumentCaptor<NewTopic> capturedNewTopic;
    @Captor
    private ArgumentCaptor<Callback<ConsumerRecord<byte[], byte[]>>> capturedConsumedCallback;
    @Captor
    private ArgumentCaptor<Callback<Void>> storeLogCallbackArgumentCaptor;

    @Before
    public void setup() throws Exception {
        Supplier<TopicAdmin> adminSupplier = () -> {
            fail("Should not attempt to instantiate admin in these tests");
            // Should never be reached; only add this thrown exception to satisfy the compiler
            throw new AssertionError();
        };
        Supplier<String> clientIdBase = () -> CLIENT_ID_BASE;

        store = spy(new KafkaOffsetBackingStore(adminSupplier, clientIdBase));

        doReturn(storeLog).when(store).createKafkaBasedLog(capturedTopic.capture(), capturedProducerProps.capture(),
                capturedConsumerProps.capture(), capturedConsumedCallback.capture(),
                capturedNewTopic.capture(), capturedAdminSupplier.capture());

        workerConfigMockedStatic = mockStatic(WorkerConfig.class, CALLS_REAL_METHODS);
        workerConfigMockedStatic.when(() -> WorkerConfig.lookupKafkaClusterId(any())).thenReturn("test-cluster");
    }

    @After
    public void tearDown() {
        workerConfigMockedStatic.close();
        verifyNoMoreInteractions(storeLog);
    }

    @Test
    public void testStartStop() {
        Map<String, String> settings = new HashMap<>(DEFAULT_PROPS);
        settings.put("offset.storage.min.insync.replicas", "3");
        settings.put("offset.storage.max.message.bytes", "1001");

        store.configure(new DistributedConfig(settings));
        store.start();

        verify(storeLog).start();

        assertEquals(TOPIC, capturedTopic.getValue());
        assertEquals("org.apache.kafka.common.serialization.ByteArraySerializer", capturedProducerProps.getValue().get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG));