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

Code before:
        expectLastCall().andThrow(new ProcessorStateException("state manager failed to close"));

        stateDirectory.unlock(taskId);

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

        assertThrows(
            ProcessorStateException.class,
            () -> StateManagerUtil.closeStateManager(
                logger, "logPrefix:", false, false, stateManager, stateDirectory, TaskType.ACTIVE));

        ctrl.verify();
    }

    @Test
    public void testCloseStateManagerWithStateStoreWipeOut() {
        expect(stateManager.taskId()).andReturn(taskId);
        expect(stateDirectory.lock(taskId)).andReturn(true);

        stateManager.close();
        expectLastCall();

        // The `baseDir` will be accessed when attempting to delete the state store.
        expect(stateManager.baseDir()).andReturn(TestUtils.tempDirectory("state_store"));

        stateDirectory.unlock(taskId);
        expectLastCall();

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

        StateManagerUtil.closeStateManager(logger,
            "logPrefix:", false, true, stateManager, stateDirectory, TaskType.ACTIVE);

        ctrl.verify();
    }

    @Test
    public void  shouldStillWipeStateStoresIfCloseThrowsException() throws IOException {
        final File randomFile = new File("/random/path");
        mockStatic(Utils.class);

        expect(stateManager.taskId()).andReturn(taskId);
        expect(stateDirectory.lock(taskId)).andReturn(true);

        stateManager.close();
        expectLastCall().andThrow(new ProcessorStateException("Close failed"));

        expect(stateManager.baseDir()).andReturn(randomFile);


Review comment (suggestion):
I do not know how much it makes sense to verify the call order of methods that should not be called.

Code after:
            utils.when(() -> Utils.delete(unknownFile)).thenThrow(new IOException("Deletion failed"));

            final ProcessorStateException thrown = assertThrows(
                    ProcessorStateException.class, () -> StateManagerUtil.closeStateManager(logger,
                            "logPrefix:", false, true, stateManager, stateDirectory, TaskType.ACTIVE));

            assertEquals(IOException.class, thrown.getCause().getClass());
        }

        inOrder.verify(stateManager).close();
        inOrder.verify(stateDirectory).unlock(taskId);
        verifyNoMoreInteractions(stateManager, stateDirectory);
    }

    @Test
    public void shouldNotCloseStateManagerIfUnableToLockTaskDirectory() {
        final InOrder inOrder = inOrder(stateManager, stateDirectory);
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateDirectory.lock(taskId)).thenReturn(false);

        StateManagerUtil.closeStateManager(
                logger, "logPrefix:", true, false, stateManager, stateDirectory, TaskType.ACTIVE);

        inOrder.verify(stateManager).taskId();
        inOrder.verify(stateDirectory).lock(taskId);
        verify(stateManager, never()).close();
        verify(stateManager, never()).baseDir();
        verify(stateDirectory, never()).unlock(taskId);
        verifyNoMoreInteractions(stateManager, stateDirectory);
    }

    @Test
    public void shouldNotWipeStateStoresIfUnableToLockTaskDirectory() {
        final InOrder inOrder = inOrder(stateManager, stateDirectory);
        when(stateManager.taskId()).thenReturn(taskId);
        when(stateDirectory.lock(taskId)).thenReturn(false);

        StateManagerUtil.closeStateManager(
                logger, "logPrefix:", false, true, stateManager, stateDirectory, TaskType.ACTIVE);

        inOrder.verify(stateManager).taskId();
        inOrder.verify(stateDirectory).lock(taskId);
        verify(stateManager, never()).close();
        verify(stateManager, never()).baseDir();
        verify(stateDirectory, never()).unlock(taskId);
        verifyNoMoreInteractions(stateManager, stateDirectory);
    }
}