Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/StateManagerUtilTest.java (Java)

Code before:
    private ProcessorTopology topology;

    @Mock(type = MockType.NICE)
    private InternalProcessorContext processorContext;

    private IMocksControl ctrl;

    private Logger logger = new LogContext("test").logger(AbstractTask.class);

    private final TaskId taskId = new TaskId(0, 0);

    @Before
    public void setup() {
        ctrl = createStrictControl();
        topology = ctrl.createMock(ProcessorTopology.class);
        processorContext = ctrl.createMock(InternalProcessorContext.class);

        stateManager = ctrl.createMock(ProcessorStateManager.class);
        stateDirectory = ctrl.createMock(StateDirectory.class);
    }

    @Test
    public void testRegisterStateStoreWhenTopologyEmpty() {
        expect(topology.stateStores()).andReturn(emptyList());

        ctrl.checkOrder(true);
        ctrl.replay();

        StateManagerUtil.registerStateStores(logger,
            "logPrefix:", topology, stateManager, stateDirectory, processorContext);

        ctrl.verify();
    }

    @Test
    public void testRegisterStateStoreFailToLockStateDirectory() {
        expect(topology.stateStores()).andReturn(singletonList(new MockKeyValueStore("store", false)));

        expect(stateManager.taskId()).andReturn(taskId);

        expect(stateDirectory.lock(taskId)).andReturn(false);

        ctrl.checkOrder(true);
        ctrl.replay();

        final LockException thrown = assertThrows(LockException.class,
            () -> StateManagerUtil.registerStateStores(logger, "logPrefix:",
                topology, stateManager, stateDirectory, processorContext));

        assertEquals("logPrefix:Failed to lock the state directory for task 0_0", thrown.getMessage());


Review comment (suggestion):
I do not think this adds much value so I removed it. If we want to ensure ordering this can be done in Mockito.

Code after:

    @Mock
    private InternalProcessorContext processorContext;

    private Logger logger = new LogContext("test").logger(AbstractTask.class);

    private final TaskId taskId = new TaskId(0, 0);

    @Test
    public void testRegisterStateStoreWhenTopologyEmpty() {
        when(topology.stateStores()).thenReturn(emptyList());

        StateManagerUtil.registerStateStores(logger,
            "logPrefix:", topology, stateManager, stateDirectory, processorContext);
    }

    @Test
    public void testRegisterStateStoreFailToLockStateDirectory() {
        when(topology.stateStores()).thenReturn(singletonList(new MockKeyValueStore("store", false)));
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateDirectory.lock(taskId)).thenReturn(false);

        final LockException thrown = assertThrows(LockException.class,
            () -> StateManagerUtil.registerStateStores(logger, "logPrefix:",
                topology, stateManager, stateDirectory, processorContext));

        assertEquals("logPrefix:Failed to lock the state directory for task 0_0", thrown.getMessage());
    }

    @Test
    public void testRegisterStateStores() {
        final MockKeyValueStore store1 = new MockKeyValueStore("store1", false);
        final MockKeyValueStore store2 = new MockKeyValueStore("store2", false);
        final List<StateStore> stateStores = Arrays.asList(store1, store2);
        final InOrder inOrder = inOrder(stateManager);
        when(topology.stateStores()).thenReturn(stateStores);
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateDirectory.lock(taskId)).thenReturn(true);
        when(stateDirectory.directoryForTaskIsEmpty(taskId)).thenReturn(true);
        when(topology.stateStores()).thenReturn(stateStores);

        StateManagerUtil.registerStateStores(logger, "logPrefix:",
            topology, stateManager, stateDirectory, processorContext);

        inOrder.verify(stateManager).registerStateStores(stateStores, processorContext);
        inOrder.verify(stateManager).initializeStoreOffsetsFromCheckpoint(true);
        verifyNoMoreInteractions(stateManager);
    }

    @Test
    public void testCloseStateManagerClean() {