<commit_before>package org.apache.ivory.service;

import org.apache.ivory.IvoryException;
import org.apache.ivory.aspect.GenericAlert;
import org.apache.ivory.entity.EntityUtil;
import org.apache.ivory.entity.v0.Entity;
import org.apache.ivory.entity.v0.Frequency;
import org.apache.ivory.entity.v0.SchemaHelper;
import org.apache.ivory.workflow.WorkflowEngineFactory;
import org.apache.ivory.workflow.engine.WorkflowEngineActionListener;
import org.apache.log4j.Logger;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

public class SLAMonitoringService implements IvoryService, WorkflowEngineActionListener {
    private static final Logger LOG = Logger.getLogger(SLAMonitoringService.class);
    public static final String SERVICE_NAME = "SLAMonitor";

    private ConcurrentMap<String, Long> monitoredEntities =
            new ConcurrentHashMap<String, Long>();

    private ConcurrentMap<String, ConcurrentMap<Date, Date>> pendingJobs =
            new ConcurrentHashMap<String, ConcurrentMap<Date, Date>>();

    private static final long INITIAL_LATENCY_SECS = 12 * 3600;

    private static final long POLL_PERIODICITY_SECS = 300;

    @Override
    public String getName() {
        return SERVICE_NAME;
    }

    @Override
    public void init() throws IvoryException {
        WorkflowEngineFactory.getWorkflowEngine().registerListener(this);
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
        executor.schedule(new Monitor(), POLL_PERIODICITY_SECS, TimeUnit.SECONDS);
    }

    @Override
    public void destroy() throws IvoryException {
    }

    @Override
    public void afterSchedule(Entity entity, String cluster) throws IvoryException {
        addEntityForMonitoring(entity, cluster);
    }

    @Override
    public void afterDelete(Entity entity, String cluster) throws IvoryException {
        removeMonitoredEntity(entity, cluster);
    }

    @Override
    public void afterSuspend(Entity entity, String cluster) throws IvoryException {
        removeMonitoredEntity(entity, cluster);
    }

    @Override
    public void afterResume(Entity entity, String cluster) throws IvoryException {
        addEntityForMonitoring(entity, cluster);
    }

    public void notifyCompletion(Entity entity, String cluster, Date nominalTime, long duration) {
        if (!isEntityMonitored(entity, cluster)) {
            addEntityForMonitoring(entity, cluster);
        }
        updateLatency(entity, cluster, duration);
        removeFromPendingList(entity, cluster, nominalTime);
    }

    private String getKey(Entity entity, String cluster) {
        return entity.toShortString() + "/" + cluster;
    }

    private void addEntityForMonitoring(Entity entity, String cluster) {
        monitoredEntities.putIfAbsent(getKey(entity, cluster), INITIAL_LATENCY_SECS);
    }

    private void removeMonitoredEntity(Entity entity, String cluster) {
        monitoredEntities.remove(getKey(entity, cluster));
        pendingJobs.remove(getKey(entity, cluster));
    }

    private boolean isEntityMonitored(Entity entity, String cluster) {
        return monitoredEntities.containsKey(getKey(entity, cluster));
    }

    private void updateLatency(Entity entity, String cluster, long duration) {
        long newLatency = (duration + monitoredEntities.get(getKey(entity, cluster))) / 2;
        monitoredEntities.put(getKey(entity, cluster), newLatency);
    }

    private void removeFromPendingList(Entity entity, String cluster, Date nominalTime) {
        ConcurrentMap<Date, Date> pendingInstances = pendingJobs.get(getKey(entity, cluster));
        LOG.debug("Removing from pending jobs: " + getKey(entity,  cluster) + " ---> " +
                SchemaHelper.formatDateUTC(nominalTime));
        pendingInstances.remove(nominalTime);
    }

    private class Monitor implements Runnable {
<commit_msg>Handling NPE as there are no pending instanes<commit_after>package org.apache.ivory.service;

import org.apache.ivory.IvoryException;
import org.apache.ivory.aspect.GenericAlert;
import org.apache.ivory.entity.EntityUtil;
import org.apache.ivory.entity.v0.Entity;
import org.apache.ivory.entity.v0.Frequency;
import org.apache.ivory.entity.v0.SchemaHelper;
import org.apache.ivory.workflow.WorkflowEngineFactory;
import org.apache.ivory.workflow.engine.WorkflowEngineActionListener;
import org.apache.log4j.Logger;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

public class SLAMonitoringService implements IvoryService, WorkflowEngineActionListener {
    private static final Logger LOG = Logger.getLogger(SLAMonitoringService.class);
    public static final String SERVICE_NAME = "SLAMonitor";

    private ConcurrentMap<String, Long> monitoredEntities =
            new ConcurrentHashMap<String, Long>();

    private ConcurrentMap<String, ConcurrentMap<Date, Date>> pendingJobs =
            new ConcurrentHashMap<String, ConcurrentMap<Date, Date>>();

    private static final long INITIAL_LATENCY_SECS = 12 * 3600;

    private static final long POLL_PERIODICITY_SECS = 300;

    @Override
    public String getName() {
        return SERVICE_NAME;
    }

    @Override
    public void init() throws IvoryException {
        WorkflowEngineFactory.getWorkflowEngine().registerListener(this);
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
        executor.schedule(new Monitor(), POLL_PERIODICITY_SECS, TimeUnit.SECONDS);
    }

    @Override
    public void destroy() throws IvoryException {
    }

    @Override
    public void afterSchedule(Entity entity, String cluster) throws IvoryException {
        addEntityForMonitoring(entity, cluster);
    }

    @Override
    public void afterDelete(Entity entity, String cluster) throws IvoryException {
        removeMonitoredEntity(entity, cluster);
    }

    @Override
    public void afterSuspend(Entity entity, String cluster) throws IvoryException {
        removeMonitoredEntity(entity, cluster);
    }

    @Override
    public void afterResume(Entity entity, String cluster) throws IvoryException {
        addEntityForMonitoring(entity, cluster);
    }

    public void notifyCompletion(Entity entity, String cluster, Date nominalTime, long duration) {
        if (!isEntityMonitored(entity, cluster)) {
            addEntityForMonitoring(entity, cluster);
        }
        updateLatency(entity, cluster, duration);
        removeFromPendingList(entity, cluster, nominalTime);
    }

    private String getKey(Entity entity, String cluster) {
        return entity.toShortString() + "/" + cluster;
    }

    private void addEntityForMonitoring(Entity entity, String cluster) {
        monitoredEntities.putIfAbsent(getKey(entity, cluster), INITIAL_LATENCY_SECS);
    }

    private void removeMonitoredEntity(Entity entity, String cluster) {
        monitoredEntities.remove(getKey(entity, cluster));
        pendingJobs.remove(getKey(entity, cluster));
    }

    private boolean isEntityMonitored(Entity entity, String cluster) {
        return monitoredEntities.containsKey(getKey(entity, cluster));
    }

    private void updateLatency(Entity entity, String cluster, long duration) {
        long newLatency = (duration + monitoredEntities.get(getKey(entity, cluster))) / 2;
        monitoredEntities.put(getKey(entity, cluster), newLatency);
    }

    private void removeFromPendingList(Entity entity, String cluster, Date nominalTime) {
        ConcurrentMap<Date, Date> pendingInstances = pendingJobs.get(getKey(entity, cluster));
        if (pendingInstances != null) {
            LOG.debug("Removing from pending jobs: " + getKey(entity,  cluster) + " ---> " +
                    SchemaHelper.formatDateUTC(nominalTime));
            pendingInstances.remove(nominalTime);
        }
    }

    private class Monitor implements Runnable {

{'fraction_non_alphanumeric': 0.08713692946058091, 'fraction_numerical': 0.00366121552355382, 'mean_word_length': 3.966666666666667, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6988697', 'n_tokens_mistral': 2287, 'n_tokens_neox': 2220, 'n_words': 512}