Drools stream integration

This passage discusses how to integrate a provided drools package into datastream application.

Packaging:
If a maven project is provided by customer. In this case, you need to ensure that the pom file contains the following:

                  org.drools        drools-bom        pom        xxx        import                         org.kie      kie-api              org.drools      drools-compiler      runtime        other dependencies                       org.kie        kie-maven-plugin        xxx        true             

In addition, a file kmodule.xml must be added to src\main\resources\META-INF folder. A minimum kmodule.xml likes like the following.

The default stateless ksession is mandatory.

Rule files can be put in main/resources as normal

The command to create jar file is still mvn package as normal. However, the jar created is a bit different. Here is a screenshot

Note that there is knowledge base cache file and kmodule file in META-INF. Two rule files in main/resources are shifted out into the root folder.

What if the customer does not provide a maven project? I guess the best strategy is to create a maven project by ourselves. If source code is provided, we just import source code into the maven project, otherwise, use customer provided jar as a maven dependency?

Note that kie module is introduced only after drools 6. So I don't think this will work for drools 5 and below. Also, for drools integration in streamtau, we are using the latest version 7.2.1. So whether earlier version like 6.x is fully compatible still remains a question.

Invocation:
Load rules:
First create a KieServices singleton instance.
private final KieServices kieServices = KieServices.Factory.get();

Load the drools package into system:

 protected DroolsDataHolder doLoadDroolsModule(DroolsLoadParam droolsLoadParam) {        DroolsParameters origParams = droolsLoadParam.getDroolsParam();        String moduleName = origParams.getModuleName();        try {            InputStream is = droolsDataLoader.getDroolsModuleAsStream(droolsLoadParam);            KieContainer curContainer = DroolsUtils.buildContainer(kieServices, is);            return new DroolsDataHolder(curContainer);        }        catch (Exception ex) {            logger.error("Error loading drools " + moduleName, ex);        }        return null;    }

DroolsDataLoader is an interface that is designed to loads drools package as stream (via either file system or restful interface)
DroolsUtils is the utility class that builds a KieContainer from stream.

public static KieContainer buildContainer(KieServices kieServices, InputStream stream) throws Exception {        Resource wrapped = kieServices.getResources().newInputStreamResource(stream);        KieModule curModule = kieServices.getRepository().addKieModule(wrapped);        ReleaseId releaseId = curModule.getReleaseId();        logger.info("Release id generated for module: {}", releaseId);        KieContainer kContainer = kieServices.newKieContainer(releaseId, DroolsUtils.class.getClassLoader());        return kContainer;    }

The returned DroolsDataHolder is merely a wrapper of KieContainer

public class DroolsDataHolder {    private final KieContainer kieContainer;    public DroolsDataHolder(KieContainer kieContainer) {        this.kieContainer = kieContainer;    }    public KieContainer getKieContainer() {        return kieContainer;    }    public void destroy() {        kieContainer.dispose();    }}

The loaded DroolsDataHolder will be cached unless rule is changed, which triggers a reload operation

public DroolsDataHolder getOrLoadDroolsModule(DroolsLoadParam droolsLoadParam) {        DroolsParameters origParams = droolsLoadParam.getDroolsParam();        String moduleName = origParams.getModuleName();        dataLock.readLock().lock();        try {            DroolsDataHolder curHolder = containers.get(moduleName);            if (curHolder != null) {                return curHolder;            }            dataLock.readLock().unlock();            dataLock.writeLock().lock();            try {                return doUpdateDroolsModule(droolsLoadParam);            }            finally {                dataLock.readLock().lock();                dataLock.writeLock().unlock();            }        }        finally {            dataLock.readLock().unlock();        }    }

Invoke the drools module:
In stream environment, only stateless drools knowledge session is supported for now. The main reason is that stream is executed in a distributed environment. The session will be created on multiple JVMS, so it is virtually hard to share all the facts globally. Evaluating the rule is quite simple, it is composed of 3 steps:

1. convert stream data to rule input pojo

   public Class getRulePojoClass(DroolsLoadParam droolsLoadParam, String inputPojoClassName) {    DroolsParameters origParams = droolsLoadParam.getDroolsParam();    String moduleName = origParams.getModuleName();    DroolsDataHolder curDataHolder = this.getOrLoadDroolsModule(droolsLoadParam);    if (curDataHolder == null) {        throw new IllegalArgumentException("No drools module found by name: " + moduleName);    }    try {        ClassLoader cl = curDataHolder.getKieContainer().getClassLoader();        Class inputPojoClass = cl.loadClass(inputPojoClassName);        return inputPojoClass;    } catch (Exception e) {        throw RtException.from(e);    }}

The good thing about drools module is that it provides a self contained class loading environment. So third party jar dependencies are unlikely to cause conflict with the outside runtime environment. However, when we build an input event to drools engine, we need to use the KieContainer's class loader to find the input event class referenced in rule.

2. build a stateless kie session and invoke the rule

   public List