Drools ConcurentCompletion 异常

如何解决Drools ConcurentCompletion 异常

我正在为一个以前没有发生过的问题而苦苦挣扎，突然它开始发生，但不知道为什么会这样。有时它发生，有时它没问题。

我收到两种不同的异常，有时给出一种，有时给出另一种。但两者似乎相关。

例外一

2021-04-02 15:28:34.848 ERROR 2948 --- [cTaskExecutor-3] c.n.o.c.c.ItemAmqpController          : java.util.concurrent.CompletionException: java.lang.NullPointerException
Caused by: java.lang.NullPointerException
at org.drools.core.util.LinkedList.remove(LinkedList.java:171)
at org.drools.core.common.DefaultAgenda.removeEagerRuleAgendaItem(DefaultAgenda.java:302)
at org.drools.core.phreak.RuleExecutor.removeRuleAgendaItemWhenEmpty(RuleExecutor.java:203)
at org.drools.core.phreak.RuleExecutor.fire(RuleExecutor.java:169)
at org.drools.core.phreak.RuleExecutor.evaluateNetworkAndFire(RuleExecutor.java:90)
at org.drools.core.concurrent.AbstractRuleEvaluator.internalEvaluateAndFire(AbstractRuleEvaluator.java:33)
at org.drools.core.concurrent.SequentialRuleEvaluator.evaluateAndFire(SequentialRuleEvaluator.java:43)
at org.drools.core.common.DefaultAgenda.fireLoop(DefaultAgenda.java:1111)
at org.drools.core.common.DefaultAgenda.internalFireAllRules(DefaultAgenda.java:1058)
at org.drools.core.common.CompositeDefaultAgenda.lambda$parallelFire$1(CompositeDefaultAgenda.java:150)
at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1604)

例外二

2021-04-02 15:26:05.017 ERROR 2948 --- [cTaskExecutor-3] c.n.o.c.c.ItemAmqpController          : java.util.concurrent.CompletionException: java.lang.indexoutofboundsexception: Index: 0,Size: 1
Caused by: java.lang.indexoutofboundsexception: Index: 0,Size: 1
at java.util.ArrayList.rangeCheck(ArrayList.java:659)
at java.util.ArrayList.get(ArrayList.java:435)
at org.drools.core.phreak.AddRemoveRule.flushLeftTupleIfNecessary(AddRemoveRule.java:627)
at org.drools.core.reteoo.LeftInputAdapterNode.doInsertSegmentMemory(LeftInputAdapterNode.java:221)
at org.drools.core.reteoo.LeftInputAdapterNode.doInsertObject(LeftInputAdapterNode.java:206)
at org.drools.core.reteoo.LeftInputAdapterNode.assertObject(LeftInputAdapterNode.java:165)
at org.drools.core.reteoo.CompositeObjectSinkAdapter.doPropagateAssertObject(CompositeObjectSinkAdapter.java:771)
at org.drools.core.reteoo.CompositeObjectSinkAdapter.propagateAssertObject(CompositeObjectSinkAdapter.java:620)
at org.drools.core.reteoo.AlphaNode.assertObject(AlphaNode.java:134)
at org.drools.core.reteoo.CompositeObjectSinkAdapter.doPropagateAssertObject(CompositeObjectSinkAdapter.java:771)
at org.drools.core.reteoo.CompositeObjectSinkAdapter.propagateAssertObject(CompositeObjectSinkAdapter.java:600)
at org.drools.core.reteoo.AlphaNode.assertObject(AlphaNode.java:134)
at org.drools.core.reteoo.CompositeObjectSinkAdapter.doPropagateAssertObject(CompositeObjectSinkAdapter.java:771)
at org.drools.core.reteoo.CompositeObjectSinkAdapter.propagateAssertObject(CompositeObjectSinkAdapter.java:620)
at org.drools.core.reteoo.CompositePartitionAwareObjectSinkAdapter$Insert.execute(CompositePartitionAwareObjectSinkAdapter.java:141)
at org.drools.core.phreak.SynchronizedPropagationList.flush(SynchronizedPropagationList.java:96)
at org.drools.core.phreak.SynchronizedPropagationList.flush(SynchronizedPropagationList.java:91)
at org.drools.core.common.DefaultAgenda.fireLoop(DefaultAgenda.java:1095)
at org.drools.core.common.DefaultAgenda.internalFireAllRules(DefaultAgenda.java:1058)
at org.drools.core.common.CompositeDefaultAgenda.lambda$parallelFire$1(CompositeDefaultAgenda.java:150)
at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1604)

错误似乎是随机发生的。用一个项目触发一个规则可能会导致这些异常，而另一个相同项目的相同规则工作得很好。这让我很难找出问题的原因

这就是会话的创建方式

private final KieSessionConfiguration configuration = KieServices.Factory.get().newKieSessionConfiguration();
configuration.setoption(ClockTypeOption.get("pseudo"));

private final KieBaseConfiguration kBaseConfig = KieServices.Factory.get().newKieBaseConfiguration();
kBaseConfig = KieServices.Factory.get().newKieBaseConfiguration();
kBaseConfig.setoption(MultithreadEvaluationoption.YES);
kBaseConfig.setoption(EqualityBehaviorOption.EQUALITY);
kBaseConfig.setoption(EventProcessingOption.STREAM);
KieBase kBase = ruleContainer.getKieContainer().newKieBase(kBaseConfig);
return kBase.newKieSession(configuration,null); 

这是我执行规则的地方

synchronized (this) { item = this.itemService.execute(item); } // Safely calling the execute method

void execute(Item item) {

        final List<KieSession> sessions = getSessions();
        final List<Item> items = itemDao.find();

        sessions
                .parallelStream()
                .forEach( kSession -> {

                    final List<Command<?>> commands = new ArrayList<Command<?>>() {{
                        add(CustomCommandFactory.newInsertElements(items));
                        add( CustomCommandFactory .newInsert(item,"item"));
                        add(new FireAllRulesCommand(agendaFilter(item)));
                    }};

                    kSession.execute(CommandFactory.newBatchExecution(commands));  // Exception thrown at here
                    kSession.dispose();
                });
}

如果您能帮助我解决这个问题，我将不胜感激。