@Override
    public void discardState() throws Exception {

        Exception aggregatedExceptions = null;

        if (jobManagerOwnedSnapshot != null) {
            try {
                jobManagerOwnedSnapshot.discardState();
            } catch (Exception remoteDiscardEx) {
                aggregatedExceptions = remoteDiscardEx;
            }
        }

        if (taskLocalSnapshot != null) {
            try {
                taskLocalSnapshot.discardState();
            } catch (Exception localDiscardEx) {
                aggregatedExceptions =
                        ExceptionUtils.firstOrSuppressed(localDiscardEx, aggregatedExceptions);
            }
        }

        if (aggregatedExceptions != null) {
            throw aggregatedExceptions;
        }
    }
@Test
    void discardState() throws Exception {
        SnapshotResult<StateObject> result =
                SnapshotResult.withLocalState(mock(StateObject.class), mock(StateObject.class));
        result.discardState();
        verify(result.getJobManagerOwnedSnapshot()).discardState();
        verify(result.getTaskLocalSnapshot()).discardState();
    }