我试着测试一个richcoflatmapfunction，我用它来做两个流的左连接，它是这样的：

private ValueState<Card> currentValueState;
    private ListState<Card> historicListState;

    @Override
    public void open(Configuration parameters) throws Exception {
        currentValueState = getRuntimeContext().getState(new ValueStateDescriptor<>("Current State", Card.class));
        historicListState = getRuntimeContext().getListState(new ListStateDescriptor<>("historic state", Card.class));
    }

    @Override
    public void flatMap1(Card currentCard, Collector<Tuple2<Card, List<Card>>> out) throws Exception {
        Iterable<Card> historicCardList =  historicListState.get();
        if (Iterables.size(historicCardList) > 0) {
            out.collect(new Tuple2<>(currentCard, Lists.newArrayList(historicCardList) ));
        } else {
            currentValueState.update(currentCard);
            out.collect(new Tuple2<>(currentCard, null));
        }
    }

    @Override
    public void flatMap2(Card historicCard, Collector<Tuple2<Card, List<Card>>> out) throws Exception {
        historicListState.add(historicCard);
    }

在flatmap1方法中，当找不到历史卡时，我将返回null

out.collect(new Tuple2<>(currentCard, null));

问题是，当我尝试测试整个功能时，我收到了以下错误：
在具有空值的候选对象上无法进行自动类型提取。请直接指定类型。
这就是我试图测试richcoflatmappfunction的方法

@Test
    public void testFlatMap() throws Exception {
        final Card current = currentCard(2L);
        final Card historic = historicCard(2L);
        final List<Card> historicList = new ArrayList<>();
        historicList.add(historic);
        CoStreamFlatMap<Card, Card, Tuple2<Card, List<Card>>> operator = new CoStreamFlatMap<>(new LeftJoin());
        KeyedTwoInputStreamOperatorTestHarness<Long, Card, Card, Tuple2<Card, List<Card>>> testHarness =
                new KeyedTwoInputStreamOperatorTestHarness<>(
                        operator,
                        (Card c) -> c.getCardHash(),
                        (Card h) -> h.getCardHash(),
                        BasicTypeInfo.LONG_TYPE_INFO);
        testHarness.setup();
        testHarness.open();
        testHarness.processElement1(new StreamRecord<>(current));
        testHarness.processElement2(new StreamRecord<>(historic));
        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
        expectedOutput.add(new StreamRecord<>(new Tuple2<>(current, historicList)));
        // Check that the result is correct
        ConcurrentLinkedQueue<Object> actualOutput = testHarness.getOutput();
        TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, actualOutput);
    }

任何帮助我会非常感谢，我有点新的apacheflink和单元测试与它谢谢。