class NiFiRemotePort extends NiFiS2S {

    @Override
    public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) {

        if (!ProvenanceEventType.SEND.equals(event.getEventType())
                && !ProvenanceEventType.RECEIVE.equals(event.getEventType())) {
            return null;
        }

        final boolean isRemoteInputPort = event.getComponentType().equals("Remote Input Port");
        final String type = isRemoteInputPort ? TYPE_NIFI_INPUT_PORT : TYPE_NIFI_OUTPUT_PORT;

        final String remotePortId = event.getComponentId();

        final S2STransitUrl s2sUrl = parseTransitURL(event.getTransitUri(), context.getClusterResolver());

        // Find connections that connects to/from the remote port.
        final List<ConnectionStatus> connections = isRemoteInputPort
                ? context.findConnectionTo(remotePortId)
                : context.findConnectionFrom(remotePortId);
        if (connections == null || connections.isEmpty()) {
            logger.warn("Connection was not found: {}", new Object[]{event});
            return null;
        }

        // The name of remote port can be retrieved from any connection, use the first one.
        final ConnectionStatus connection = connections.get(0);
        final Referenceable ref = new Referenceable(type);
        ref.set(ATTR_NAME, isRemoteInputPort ? connection.getDestinationName() : connection.getSourceName());
        ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(s2sUrl.clusterName, s2sUrl.targetPortId));

        return singleDataSetRef(event.getComponentId(), event.getEventType(), ref);
    }

}