Drools OOM

LargeTestFact.java

package org.drools.compiler.integrationtests;

import java.util.Arrays;
import java.util.Date;
import java.util.Objects;

public class LargeTestFact {

    private final long id;
    private final Date timestamp;
    private final byte[] data;

    public LargeTestFact(long id, Date timestamp, int size) {
        this.id = id;
        this.timestamp = Objects.requireNonNull(timestamp);
        data = new byte[size];
    }

    public long getId() {
        return id;
    }

    public Date getTimestamp() {
        return timestamp;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof LargeTestFact)) return false;
        LargeTestFact that = (LargeTestFact) o;
        return id == that.id &&
                Objects.equals(timestamp, that.timestamp) &&
                Arrays.equals(data, that.data);
    }

    @Override
    public int hashCode() {
        return Objects.hash(id, timestamp);
    }

    @Override
    public String toString() {
        return "LargeTestFact{" +
                "id=" + id +
                ", timestamp=" + timestamp +
                '}';
    }
}

WhatsWrongWithMeTest.java

/*******************************************************************************
 * This file is part of OpenNMS(R).
 *
 * Copyright (C) 2019 The OpenNMS Group, Inc.
 * OpenNMS(R) is Copyright (C) 1999-2019 The OpenNMS Group, Inc.
 *
 * OpenNMS(R) is a registered trademark of The OpenNMS Group, Inc.
 *
 * OpenNMS(R) is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published
 * by the Free Software Foundation, either version 3 of the License,
 * or (at your option) any later version.
 *
 * OpenNMS(R) is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with OpenNMS(R).  If not, see:
 *      http://www.gnu.org/licenses/
 *
 * For more information contact:
 *     OpenNMS(R) Licensing <license@opennms.org>
 *     http://www.opennms.org/
 *     http://www.opennms.com/
 *******************************************************************************/

package org.drools.compiler.integrationtests;

import static org.drools.compiler.CommonTestMethodBase.createAndDeployJar;

import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.drools.core.RuleBaseConfiguration;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.kie.api.KieBase;
import org.kie.api.KieServices;
import org.kie.api.builder.KieModule;
import org.kie.api.builder.ReleaseId;
import org.kie.api.conf.EventProcessingOption;
import org.kie.api.runtime.KieContainer;
import org.kie.api.runtime.KieSession;
import org.kie.api.runtime.rule.FactHandle;

import com.google.common.util.concurrent.RateLimiter;

public class WhatsWrongWithMeTest {

    @Rule
    public TemporaryFolder temporaryFolder = new TemporaryFolder();

    @Test
    public void canOOM() throws InterruptedException {
        KieServices ks = KieServices.Factory.get();
        // Create an in-memory jar for version 1.0.0
        ReleaseId releaseId1 = ks.newReleaseId( "org.kie", "test-testIt", "1.0" );
        KieModule km = createAndDeployJar( ks, releaseId1, "package org.drools.compiler\n" +
                "declare org.drools.compiler.integrationtests.LargeTestFact\n" +
                "@role(event)\n" +
                "@timestamp(timestamp)\n" +
                "end\n" +
                "rule 'delete outside of window' when\n" +
                "   $fact : LargeTestFact( )\n" +
                "   not( LargeTestFact( this == $fact ) over window:time( 8h ) )\n" +
                "then\n" +
                "    retract($fact);\n" +
                "end\n");

        // Create a session and fire rules
        KieContainer kc = ks.newKieContainer( km.getReleaseId() );

        // Create KieBase
        RuleBaseConfiguration ruleBaseConfig = new RuleBaseConfiguration();
        ruleBaseConfig.setEventProcessingMode(EventProcessingOption.STREAM);
        KieBase base = kc.newKieBase(ruleBaseConfig);

        // Create a dedicated thread to fire
        KieSession kieSession = base.newKieSession();
        CountDownLatch latch = new CountDownLatch(1);
        Thread t = new Thread(() -> {
            latch.countDown();
            System.out.println("Firing.");
            kieSession.fireUntilHalt();
            System.out.println("Halted.");
        });
        t.start();

        // Wait for the thread to start firing
        latch.await();
        Thread.sleep(250);

        // Increase delta to add variations in the timestamp - creating out-of-order events
        Random random = new Random(0);
        long delta = TimeUnit.DAYS.toMillis(30);

        // # of facts kept in the session at any given time
        int numFactsInFlight = 100;
        List<FactHandle> factsInFlight = new LinkedList<>();

        // # of facts to process before trying to acquire from the rate limiter
        int batchSize = 1000;

        // size of byte array used in the fact
        int factByteSize = 1024;

        final RateLimiter rateLimiter = RateLimiter.create(100000);
        long i = 0;
        while(true) {
            rateLimiter.acquire(batchSize);
            for (int j = 0; j < batchSize; j++) {
                Date effectiveDate = new Date(System.currentTimeMillis() + (long)Math.floor(random.nextDouble() * delta));
                LargeTestFact fact = new LargeTestFact(i++, effectiveDate, factByteSize);
                factsInFlight.add(kieSession.insert(fact));
                //System.out.printf("Inserting fact with timestamp: %d\n", effectiveDate.getTime());
                if (factsInFlight.size() > numFactsInFlight) {
                    kieSession.delete(factsInFlight.remove(0));
                    //System.out.println("Deleting fact.");
                }
            }
        }
    }

}