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

Code before:
    }
    private static final Map<ByteBuffer, ByteBuffer> FIRST_SET = new HashMap<>();
    static {
        FIRST_SET.put(buffer("key"), buffer("value"));
        FIRST_SET.put(null, null);
    }

    private static final ByteBuffer TP0_KEY = buffer("TP0KEY");
    private static final ByteBuffer TP1_KEY = buffer("TP1KEY");
    private static final ByteBuffer TP2_KEY = buffer("TP2KEY");
    private static final ByteBuffer TP0_VALUE = buffer("VAL0");
    private static final ByteBuffer TP1_VALUE = buffer("VAL1");
    private static final ByteBuffer TP2_VALUE = buffer("VAL2");
    private static final ByteBuffer TP0_VALUE_NEW = buffer("VAL0_NEW");
    private static final ByteBuffer TP1_VALUE_NEW = buffer("VAL1_NEW");

    @Mock
    KafkaBasedLog<byte[], byte[]> storeLog;
    private KafkaOffsetBackingStore store;

    private Capture<String> capturedTopic = EasyMock.newCapture();
    private Capture<Map<String, Object>> capturedProducerProps = EasyMock.newCapture();
    private Capture<Map<String, Object>> capturedConsumerProps = EasyMock.newCapture();
    private Capture<Supplier<TopicAdmin>> capturedAdminSupplier = EasyMock.newCapture();
    private Capture<NewTopic> capturedNewTopic = EasyMock.newCapture();
    private Capture<Callback<ConsumerRecord<byte[], byte[]>>> capturedConsumedCallback = EasyMock.newCapture();

    @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();


Review comment (refactor):
I usually like annotations when they simplify things, but `@InjectMocks` is a bit too magical for my taste, and it restricts us from being able to pass in non-mocked and non-null constructor arguments. I also don't see it in use anywhere else in our tests.

Could we replace this with some logic in `setup()` to explicitly construct the `store`?

Code after:
        DEFAULT_PROPS.put(DistributedConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
        DEFAULT_PROPS.put(DistributedConfig.VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.json.JsonConverter");
        DEFAULT_DISTRIBUTED_CONFIG = new DistributedConfig(DEFAULT_PROPS);

    }
    private static final Map<ByteBuffer, ByteBuffer> FIRST_SET = new HashMap<>();
    static {
        FIRST_SET.put(buffer("key"), buffer("value"));
        FIRST_SET.put(null, null);
    }

    private static final ByteBuffer TP0_KEY = buffer("TP0KEY");
    private static final ByteBuffer TP1_KEY = buffer("TP1KEY");
    private static final ByteBuffer TP2_KEY = buffer("TP2KEY");
    private static final ByteBuffer TP0_VALUE = buffer("VAL0");
    private static final ByteBuffer TP1_VALUE = buffer("VAL1");
    private static final ByteBuffer TP2_VALUE = buffer("VAL2");
    private static final ByteBuffer TP0_VALUE_NEW = buffer("VAL0_NEW");
    private static final ByteBuffer TP1_VALUE_NEW = buffer("VAL1_NEW");

    @Mock
    KafkaBasedLog<byte[], byte[]> storeLog;
    private KafkaOffsetBackingStore store;

    private MockedStatic<WorkerConfig> workerConfigMockedStatic;

    @Captor
    private ArgumentCaptor<String> capturedTopic;
    @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));