@Test public void testRemoteOutputPort() { final String componentType = "Remote Output Port"; final String transitUri = "http: final ProvenanceEventRecord record = Mockito.mock(ProvenanceEventRecord.class); when(record.getComponentId()).thenReturn("port-guid"); when(record.getComponentType()).thenReturn(componentType); when(record.getTransitUri()).thenReturn(transitUri); when(record.getEventType()).thenReturn(ProvenanceEventType.RECEIVE); final ClusterResolvers clusterResolvers = Mockito.mock(ClusterResolvers.class); when(clusterResolvers.fromHostNames(matches(".+\\.example\\.com"))).thenReturn("cluster1"); final List<ConnectionStatus> connections = new ArrayList<>(); final ConnectionStatus connection = new ConnectionStatus(); connection.setSourceId("port-guid"); connection.setSourceName("outputPortA"); connections.add(connection); final AnalysisContext context = Mockito.mock(AnalysisContext.class); when(context.getClusterResolver()).thenReturn(clusterResolvers); when(context.findConnectionFrom(matches("port-guid"))).thenReturn(connections); final NiFiProvenanceEventAnalyzer analyzer = NiFiProvenanceEventAnalyzerFactory.getAnalyzer(componentType, transitUri, record.getEventType()); assertNotNull(analyzer); final DataSetRefs refs = analyzer.analyze(context, record); assertEquals(1, refs.getInputs().size()); assertEquals(0, refs.getOutputs().size()); Referenceable ref = refs.getInputs().iterator().next(); assertEquals(TYPE_NIFI_OUTPUT_PORT, ref.getTypeName()); assertEquals("outputPortA", ref.get(ATTR_NAME)); assertEquals("port-guid@cluster1", ref.get(ATTR_QUALIFIED_NAME)); }
@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()); 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; } 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); }
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()); 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; } 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); } }
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()); 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; } 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); }  }
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()); 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; } 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); }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetComponentTypePattern(); }
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()); 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; } 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); }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetComponentTypePattern();  }