@Test public void testConsumeKafka() { final String processorName = "ConsumeKafka"; final String transitUri = "PLAINTEXT: final ProvenanceEventRecord record = Mockito.mock(ProvenanceEventRecord.class); when(record.getComponentType()).thenReturn(processorName); 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 AnalysisContext context = Mockito.mock(AnalysisContext.class); when(context.getClusterResolver()).thenReturn(clusterResolvers); final NiFiProvenanceEventAnalyzer analyzer = NiFiProvenanceEventAnalyzerFactory.getAnalyzer(processorName, 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("kafka_topic", ref.getTypeName()); assertEquals("topicA", ref.get(ATTR_NAME)); assertEquals("topicA", ref.get("topic")); assertEquals("topicA@cluster1", ref.get(ATTR_QUALIFIED_NAME)); }
@Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final String transitUri = event.getTransitUri(); if (transitUri == null) { return null; } final Matcher uriMatcher = URI_PATTERN.matcher(transitUri); if (!uriMatcher.matches()) { logger.warn("Unexpected transit URI: {}", new Object[]{transitUri}); return null; } final String[] hostNames = splitHostNames(uriMatcher.group(1)); final String clusterName = context.getClusterResolver().fromHostNames(hostNames); final String topicName = uriMatcher.group(2); ref.set(ATTR_NAME, topicName); ref.set(ATTR_TOPIC, topicName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, topicName)); ref.set(ATTR_URI, transitUri); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }
KafkaTopic extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final String transitUri = event.getTransitUri(); if (transitUri == null) { return null; } final Matcher uriMatcher = URI_PATTERN.matcher(transitUri); if (!uriMatcher.matches()) { logger.warn("Unexpected transit URI: {}", new Object[]{transitUri}); return null; } final String[] hostNames = splitHostNames(uriMatcher.group(1)); final String clusterName = context.getClusterResolver().fromHostNames(hostNames); final String topicName = uriMatcher.group(2); ref.set(ATTR_NAME, topicName); ref.set(ATTR_TOPIC, topicName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, topicName)); ref.set(ATTR_URI, transitUri); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); } }
KafkaTopic extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final String transitUri = event.getTransitUri(); if (transitUri == null) { return null; } final Matcher uriMatcher = URI_PATTERN.matcher(transitUri); if (!uriMatcher.matches()) { logger.warn("Unexpected transit URI: {}", new Object[]{transitUri}); return null; } final String[] hostNames = splitHostNames(uriMatcher.group(1)); final String clusterName = context.getClusterResolver().fromHostNames(hostNames); final String topicName = uriMatcher.group(2); ref.set(ATTR_NAME, topicName); ref.set(ATTR_TOPIC, topicName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, topicName)); ref.set(ATTR_URI, transitUri); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }  }
KafkaTopic extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final String transitUri = event.getTransitUri(); if (transitUri == null) { return null; } final Matcher uriMatcher = URI_PATTERN.matcher(transitUri); if (!uriMatcher.matches()) { logger.warn("Unexpected transit URI: {}", new Object[]{transitUri}); return null; } final String[] hostNames = splitHostNames(uriMatcher.group(1)); final String clusterName = context.getClusterResolver().fromHostNames(hostNames); final String topicName = uriMatcher.group(2); ref.set(ATTR_NAME, topicName); ref.set(ATTR_TOPIC, topicName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, topicName)); ref.set(ATTR_URI, transitUri); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetComponentTypePattern(); }
KafkaTopic extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final String transitUri = event.getTransitUri(); if (transitUri == null) { return null; } final Matcher uriMatcher = URI_PATTERN.matcher(transitUri); if (!uriMatcher.matches()) { logger.warn("Unexpected transit URI: {}", new Object[]{transitUri}); return null; } final String[] hostNames = splitHostNames(uriMatcher.group(1)); final String clusterName = context.getClusterResolver().fromHostNames(hostNames); final String topicName = uriMatcher.group(2); ref.set(ATTR_NAME, topicName); ref.set(ATTR_TOPIC, topicName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, topicName)); ref.set(ATTR_URI, transitUri); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetComponentTypePattern();  }