@Test public void testHDFSPath() { final String processorName = "PutHDFS"; final String transitUri = "hdfs: final ProvenanceEventRecord record = Mockito.mock(ProvenanceEventRecord.class); when(record.getComponentType()).thenReturn(processorName); when(record.getTransitUri()).thenReturn(transitUri); when(record.getEventType()).thenReturn(ProvenanceEventType.SEND); 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(0, refs.getInputs().size()); assertEquals(1, refs.getOutputs().size()); Referenceable ref = refs.getOutputs().iterator().next(); assertEquals("hdfs_path", ref.getTypeName()); assertEquals("/user/nifi/fileA", ref.get(ATTR_NAME)); assertEquals("/user/nifi/fileA@cluster1", ref.get(ATTR_QUALIFIED_NAME)); }
@Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final String path = uri.getPath(); ref.set(ATTR_NAME, path); ref.set(ATTR_PATH, path); ref.set(ATTR_CLUSTER_NAME, clusterName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, path)); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }
HDFSPath extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final String path = uri.getPath(); ref.set(ATTR_NAME, path); ref.set(ATTR_PATH, path); ref.set(ATTR_CLUSTER_NAME, clusterName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, path)); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); } }
HDFSPath extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final String path = uri.getPath(); ref.set(ATTR_NAME, path); ref.set(ATTR_PATH, path); ref.set(ATTR_CLUSTER_NAME, clusterName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, path)); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }  }
HDFSPath extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final String path = uri.getPath(); ref.set(ATTR_NAME, path); ref.set(ATTR_PATH, path); ref.set(ATTR_CLUSTER_NAME, clusterName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, path)); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetTransitUriPattern(); }
HDFSPath extends AbstractNiFiProvenanceEventAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { final Referenceable ref = new Referenceable(TYPE); final URI uri = parseUri(event.getTransitUri()); final String clusterName = context.getClusterResolver().fromHostNames(uri.getHost()); final String path = uri.getPath(); ref.set(ATTR_NAME, path); ref.set(ATTR_PATH, path); ref.set(ATTR_CLUSTER_NAME, clusterName); ref.set(ATTR_QUALIFIED_NAME, toQualifiedName(clusterName, path)); return singleDataSetRef(event.getComponentId(), event.getEventType(), ref); }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetTransitUriPattern();  }