class SiteToSiteMetricsReportingTask 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 VirtualMachineMetrics virtualMachineMetrics = VirtualMachineMetrics.getInstance();
        final Map<String, ?> config = Collections.emptyMap();
        final JsonBuilderFactory factory = Json.createBuilderFactory(config);

        final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue();
        final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue();
        final ProcessGroupStatus status = context.getEventAccess().getControllerStatus();

        if(status != null) {
            final Map<String,String> statusMetrics = metricsService.getMetrics(status, false);
            final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics);

            final MetricsBuilder metricsBuilder = new MetricsBuilder(factory);
            final OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean();
            final double systemLoad = os.getSystemLoadAverage();

            byte[] data;
            final Map<String, String> attributes = new HashMap<>();

            if(context.getProperty(FORMAT).getValue().equals(AMBARI_FORMAT.getValue())) {
                final JsonObject metricsObject = metricsBuilder
                        .applicationId(applicationId)
                        .instanceId(status.getId())
                        .hostname(hostname)
                        .timestamp(System.currentTimeMillis())
                        .addAllMetrics(statusMetrics)
                        .addAllMetrics(jvmMetrics)
                        .metric(MetricNames.CORES, String.valueOf(os.getAvailableProcessors()))
                        .metric(MetricNames.LOAD1MN, String.valueOf(systemLoad >= 0 ? systemLoad : -1))
                        .build();

                data = metricsObject.toString().getBytes(StandardCharsets.UTF_8);
                attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json");
            } else {
                final JsonObject metricsObject = metricsService.getMetrics(factory, status, virtualMachineMetrics, applicationId, status.getId(),
                        hostname, System.currentTimeMillis(), os.getAvailableProcessors(), systemLoad >= 0 ? systemLoad : -1);
                data = getData(context, new ByteArrayInputStream(metricsObject.toString().getBytes(StandardCharsets.UTF_8)), attributes);
            }

            try {
                long start = System.nanoTime();
                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 String transactionId = UUID.randomUUID().toString();
                attributes.put("reporting.task.transaction.id", transactionId);
                attributes.put("reporting.task.name", getName());
                attributes.put("reporting.task.uuid", getIdentifier());
                attributes.put("reporting.task.type", this.getClass().getSimpleName());

                transaction.send(data, attributes);
                transaction.confirm();
                transaction.complete();

                final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
                getLogger().info("Successfully sent metrics to destination in {}ms; Transaction ID = {}", new Object[]{transferMillis, transactionId});
            } catch (final Exception e) {
                throw new ProcessException("Failed to send metrics to destination due to:" + e.getMessage(), e);
            }

        } else {
            getLogger().error("No process group status to retrieve metrics");
        }
    }

    public  SiteToSiteMetricsReportingTask();

}