@Test public void testWhenProvenanceMaxIdEqualToLastEventIdInStateManager() throws IOException, InitializationException { final long maxEventId = 2500; final MockSiteToSiteProvenanceReportingTask task = new MockSiteToSiteProvenanceReportingTask(); final MockStateManager stateManager = new MockStateManager(task); final Map<String,String> state = new HashMap<>(); state.put(SiteToSiteProvenanceReportingTask.LAST_EVENT_ID_KEY, String.valueOf(maxEventId)); stateManager.setState(state, Scope.LOCAL); final ReportingContext context = Mockito.mock(ReportingContext.class); Mockito.when(context.getStateManager()).thenReturn(stateManager); final ProvenanceEventRepository provenanceRepository = Mockito.mock(ProvenanceEventRepository.class); Mockito.doAnswer(new Answer<Long>() { @Override public Long answer(final InvocationOnMock invocation) throws Throwable { return maxEventId; } }).when(provenanceRepository).getMaxEventId(); final EventAccess eventAccess = Mockito.mock(EventAccess.class); Mockito.when(context.getEventAccess()).thenReturn(eventAccess); Mockito.when(eventAccess.getProvenanceRepository()).thenReturn(provenanceRepository); final ComponentLog logger = Mockito.mock(ComponentLog.class); final ReportingInitializationContext initContext = Mockito.mock(ReportingInitializationContext.class); Mockito.when(initContext.getIdentifier()).thenReturn(UUID.randomUUID().toString()); Mockito.when(initContext.getLogger()).thenReturn(logger); task.initialize(initContext); task.onTrigger(context); assertEquals(0, task.dataSent.size()); }
@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) { 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(); 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(); 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); } 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; 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; } } }
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) { 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(); 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(); 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); } 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; 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; } } } }
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) { 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(); 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(); 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); } 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; 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; } } }  }
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) { 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(); 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(); 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); } 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; 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; } } }  @Override void onTrigger(final ReportingContext context); }
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) { 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(); 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(); 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); } 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; 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; } } }  @Override void onTrigger(final ReportingContext context);  }