class SiteToSiteProvenanceReportingTask extends AbstractSiteToSiteReportingTask {

    @Override
    public void onTrigger(final ReportingContext context) {
        final boolean isClustered = context.isClustered();
        final String nodeId = context.getClusterNodeIdentifier();
        if (nodeId == null && isClustered) {
            getLogger().debug("This instance of NiFi is configured for clustering, but the Cluster Node Identifier is not yet available. "
                + "Will wait for Node Identifier to be established.");
            return;
        }

        final ProcessGroupStatus procGroupStatus = context.getEventAccess().getControllerStatus();
        final String rootGroupName = procGroupStatus == null ? null : procGroupStatus.getName();
        final Map<String,String> componentMap = createComponentMap(procGroupStatus);

        Long currMaxId = context.getEventAccess().getProvenanceRepository().getMaxEventId();

        if(currMaxId == null) {
            getLogger().debug("No events to send because no events have been created yet.");
            return;
        }

        if (firstEventId < 0) {
            Map<String, String> state;
            try {
                state = context.getStateManager().getState(Scope.LOCAL).toMap();
            } catch (IOException e) {
                getLogger().error("Failed to get state at start up due to:" + e.getMessage(), e);
                return;
            }
            if (state.containsKey(LAST_EVENT_ID_KEY)) {
                firstEventId = Long.parseLong(state.get(LAST_EVENT_ID_KEY)) + 1;
            }

            if(currMaxId < (firstEventId - 1)){
                getLogger().warn("Current provenance max id is {} which is less than what was stored in state as the last queried event, which was {}. This means the provenance restarted its " +
                        "ids. Restarting querying from the beginning.", new Object[]{currMaxId, firstEventId});
                firstEventId = -1;
            }
        }

        if (currMaxId == (firstEventId - 1)) {
            getLogger().debug("No events to send due to the current max id being equal to the last id that was queried.");
            return;
        }

        List<ProvenanceEventRecord> events;
        try {
            events = context.getEventAccess().getProvenanceEvents(firstEventId, context.getProperty(BATCH_SIZE).asInteger());
        } catch (final IOException ioe) {
            getLogger().error("Failed to retrieve Provenance Events from repository due to: " + ioe.getMessage(), ioe);
            return;
        }

        if (events == null || events.isEmpty()) {
            getLogger().debug("No events to send due to 'events' being null or empty.");
            return;
        }

        final String nifiUrl = context.getProperty(INSTANCE_URL).evaluateAttributeExpressions().getValue();
        URL url;
        try {
            url = new URL(nifiUrl);
        } catch (final MalformedURLException e1) {
            // already validated
            throw new AssertionError();
        }

        final String hostname = url.getHost();
        final String platform = context.getProperty(PLATFORM).evaluateAttributeExpressions().getValue();

        final Map<String, ?> config = Collections.emptyMap();
        final JsonBuilderFactory factory = Json.createBuilderFactory(config);
        final JsonObjectBuilder builder = factory.createObjectBuilder();

        final DateFormat df = new SimpleDateFormat(TIMESTAMP_FORMAT);
        df.setTimeZone(TimeZone.getTimeZone("Z"));

        while (events != null && !events.isEmpty()) {
            final long start = System.nanoTime();

            // Create a JSON array of all the events in the current batch
            final JsonArrayBuilder arrayBuilder = factory.createArrayBuilder();
            for (final ProvenanceEventRecord event : events) {
                final String componentName = componentMap.get(event.getComponentId());
                arrayBuilder.add(serialize(factory, builder, event, df, componentName, hostname, url, rootGroupName, platform, nodeId));
            }
            final JsonArray jsonArray = arrayBuilder.build();

            // Send the JSON document for the current batch
            try {
                final Transaction transaction = getClient().createTransaction(TransferDirection.SEND);
                if (transaction == null) {
                    getLogger().debug("All destination nodes are penalized; will attempt to send data later");
                    return;
                }

                final Map<String, String> attributes = new HashMap<>();
                final String transactionId = UUID.randomUUID().toString();
                attributes.put("reporting.task.transaction.id", transactionId);

                final byte[] data = jsonArray.toString().getBytes(StandardCharsets.UTF_8);
                transaction.send(data, attributes);
                transaction.confirm();
                transaction.complete();

                final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
                getLogger().info("Successfully sent {} Provenance Events to destination in {} ms; Transaction ID = {}; First Event ID = {}",
                        new Object[]{events.size(), transferMillis, transactionId, events.get(0).getEventId()});
            } catch (final IOException e) {
                throw new ProcessException("Failed to send Provenance Events to destination due to IOException:" + e.getMessage(), e);
            }

            // Store the id of the last event so we know where we left off
            final ProvenanceEventRecord lastEvent = events.get(events.size() - 1);
            final String lastEventId = String.valueOf(lastEvent.getEventId());
            try {
                StateManager stateManager = context.getStateManager();
                Map<String, String> newMapOfState = new HashMap<>();
                newMapOfState.put(LAST_EVENT_ID_KEY, lastEventId);
                stateManager.setState(newMapOfState, Scope.LOCAL);
            } catch (final IOException ioe) {
                getLogger().error("Failed to update state to {} due to {}; this could result in events being re-sent after a restart. The message of {} was: {}",
                        new Object[]{lastEventId, ioe, ioe, ioe.getMessage()}, ioe);
            }

            firstEventId = lastEvent.getEventId() + 1;

            // Retrieve the next batch
            try {
                events = context.getEventAccess().getProvenanceEvents(firstEventId, context.getProperty(BATCH_SIZE).asInteger());
            } catch (final IOException ioe) {
                getLogger().error("Failed to retrieve Provenance Events from repository due to: " + ioe.getMessage(), ioe);
                return;
            }
        }

    }

}