@Test public void testTableLineageWithZookeeperDiscovery() { final String processorName = "PutHiveQL"; final String transitUri = "jdbc:hive2: final ProvenanceEventRecord record = Mockito.mock(ProvenanceEventRecord.class); when(record.getComponentType()).thenReturn(processorName); when(record.getTransitUri()).thenReturn(transitUri); when(record.getEventType()).thenReturn(ProvenanceEventType.SEND); when(record.getAttribute(ATTR_INPUT_TABLES)).thenReturn("tableA1, tableA2"); when(record.getAttribute(ATTR_OUTPUT_TABLES)).thenReturn("databaseB.tableB1"); final ClusterResolvers clusterResolvers = Mockito.mock(ClusterResolvers.class); when(clusterResolvers.fromHostNames(eq("0.example.com"), eq("1.example.com"), eq("2.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(2, refs.getInputs().size()); final Map<String, String> expectedInputRefs = new HashMap<>(); expectedInputRefs.put("default.tableA1@cluster1", "tableA1"); expectedInputRefs.put("default.tableA2@cluster1", "tableA2"); for (Referenceable ref : refs.getInputs()) { final String qName = (String) ref.get(ATTR_QUALIFIED_NAME); assertTrue(expectedInputRefs.containsKey(qName)); assertEquals(expectedInputRefs.get(qName), ref.get(ATTR_NAME)); } assertEquals(1, refs.getOutputs().size()); Referenceable ref = refs.getOutputs().iterator().next(); assertEquals("hive_table", ref.getTypeName()); assertEquals("tableB1", ref.get(ATTR_NAME)); assertEquals("databaseB.tableB1@cluster1", ref.get(ATTR_QUALIFIED_NAME)); }
@Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { 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 clusterName = context.getClusterResolver().fromHostNames(splitHostNames(uriMatcher.group(1))); String connectedDatabaseName = null; if (uriMatcher.groupCount() > 1) { final String[] connectionParams = uriMatcher.group(2).split(";"); connectedDatabaseName = connectionParams[0]; } if (StringUtils.isEmpty(connectedDatabaseName)) { connectedDatabaseName = "default"; } final Set<Tuple<String, String>> inputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_INPUT_TABLES)); final Set<Tuple<String, String>> outputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_OUTPUT_TABLES)); if (inputTables.isEmpty() && outputTables.isEmpty()) { return getDatabaseRef(event.getComponentId(), event.getEventType(), clusterName, connectedDatabaseName); } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); addRefs(refs, true, clusterName, inputTables); addRefs(refs, false, clusterName, outputTables); return refs; }
Hive2JDBC extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { 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 clusterName = context.getClusterResolver().fromHostNames(splitHostNames(uriMatcher.group(1))); String connectedDatabaseName = null; if (uriMatcher.groupCount() > 1) { final String[] connectionParams = uriMatcher.group(2).split(";"); connectedDatabaseName = connectionParams[0]; } if (StringUtils.isEmpty(connectedDatabaseName)) { connectedDatabaseName = "default"; } final Set<Tuple<String, String>> inputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_INPUT_TABLES)); final Set<Tuple<String, String>> outputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_OUTPUT_TABLES)); if (inputTables.isEmpty() && outputTables.isEmpty()) { return getDatabaseRef(event.getComponentId(), event.getEventType(), clusterName, connectedDatabaseName); } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); addRefs(refs, true, clusterName, inputTables); addRefs(refs, false, clusterName, outputTables); return refs; } }
Hive2JDBC extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { 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 clusterName = context.getClusterResolver().fromHostNames(splitHostNames(uriMatcher.group(1))); String connectedDatabaseName = null; if (uriMatcher.groupCount() > 1) { final String[] connectionParams = uriMatcher.group(2).split(";"); connectedDatabaseName = connectionParams[0]; } if (StringUtils.isEmpty(connectedDatabaseName)) { connectedDatabaseName = "default"; } final Set<Tuple<String, String>> inputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_INPUT_TABLES)); final Set<Tuple<String, String>> outputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_OUTPUT_TABLES)); if (inputTables.isEmpty() && outputTables.isEmpty()) { return getDatabaseRef(event.getComponentId(), event.getEventType(), clusterName, connectedDatabaseName); } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); addRefs(refs, true, clusterName, inputTables); addRefs(refs, false, clusterName, outputTables); return refs; }  }
Hive2JDBC extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { 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 clusterName = context.getClusterResolver().fromHostNames(splitHostNames(uriMatcher.group(1))); String connectedDatabaseName = null; if (uriMatcher.groupCount() > 1) { final String[] connectionParams = uriMatcher.group(2).split(";"); connectedDatabaseName = connectionParams[0]; } if (StringUtils.isEmpty(connectedDatabaseName)) { connectedDatabaseName = "default"; } final Set<Tuple<String, String>> inputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_INPUT_TABLES)); final Set<Tuple<String, String>> outputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_OUTPUT_TABLES)); if (inputTables.isEmpty() && outputTables.isEmpty()) { return getDatabaseRef(event.getComponentId(), event.getEventType(), clusterName, connectedDatabaseName); } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); addRefs(refs, true, clusterName, inputTables); addRefs(refs, false, clusterName, outputTables); return refs; }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetTransitUriPattern(); }
Hive2JDBC extends AbstractHiveAnalyzer { @Override public DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event) { 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 clusterName = context.getClusterResolver().fromHostNames(splitHostNames(uriMatcher.group(1))); String connectedDatabaseName = null; if (uriMatcher.groupCount() > 1) { final String[] connectionParams = uriMatcher.group(2).split(";"); connectedDatabaseName = connectionParams[0]; } if (StringUtils.isEmpty(connectedDatabaseName)) { connectedDatabaseName = "default"; } final Set<Tuple<String, String>> inputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_INPUT_TABLES)); final Set<Tuple<String, String>> outputTables = parseTableNames(connectedDatabaseName, event.getAttribute(ATTR_OUTPUT_TABLES)); if (inputTables.isEmpty() && outputTables.isEmpty()) { return getDatabaseRef(event.getComponentId(), event.getEventType(), clusterName, connectedDatabaseName); } final DataSetRefs refs = new DataSetRefs(event.getComponentId()); addRefs(refs, true, clusterName, inputTables); addRefs(refs, false, clusterName, outputTables); return refs; }  @Override DataSetRefs analyze(AnalysisContext context, ProvenanceEventRecord event); @Override String targetTransitUriPattern();  }