@Test void testSourceLineageInfo() { SourceState sourceState = new SourceState(); sourceState.setProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY, "salesforce"); sourceState.setProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY, "snapshot_append"); sourceState.setProp(Partitioner.HAS_USER_SPECIFIED_PARTITIONS, true); sourceState.setProp(Partitioner.USER_SPECIFIED_PARTITIONS, "20140213000000,20170407152123"); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_EXTRACT_TYPE, "SNAPSHOT"); QueryBasedSource.SourceEntity sourceEntity = QueryBasedSource.SourceEntity.fromSourceEntityName("contacts"); SalesforceSource source = new SalesforceSource(new LineageInfo(ConfigFactory.empty())); List<WorkUnit> workUnits = source.generateWorkUnits(sourceEntity, sourceState, 20140213000000L); Assert.assertEquals(workUnits.size(), 1); String expected = "{\"object-type\":\"org.apache.gobblin.dataset.DatasetDescriptor\"," + "\"object-data\":{\"platform\":\"salesforce\",\"metadata\":{},\"name\":\"contacts\"}}"; Assert.assertEquals(expected, workUnits.get(0).getProp("gobblin.event.lineage.source")); Assert.assertEquals(workUnits.get(0).getProp("gobblin.event.lineage.name"), "contacts"); }
@Override protected List<WorkUnit> generateWorkUnits(SourceEntity sourceEntity, SourceState state, long previousWatermark) { List<WorkUnit> workUnits = null; workUnitConf = new SfConfig(state.getProperties()); String partitionType = state.getProp(SALESFORCE_PARTITION_TYPE, ""); if (partitionType.equals("PK_CHUNKING")) { workUnits = generateWorkUnitsPkChunking(sourceEntity, state, previousWatermark); } else { workUnits = generateWorkUnitsStrategy(sourceEntity, state, previousWatermark); } log.info("====Generated {} workUnit(s)====", workUnits.size()); if (workUnitConf.partitionOnly) { log.info("It is partitionOnly mode, return blank workUnit list"); return new ArrayList<>(); } else { return workUnits; } }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { @Override protected List<WorkUnit> generateWorkUnits(SourceEntity sourceEntity, SourceState state, long previousWatermark) { List<WorkUnit> workUnits = null; workUnitConf = new SfConfig(state.getProperties()); String partitionType = state.getProp(SALESFORCE_PARTITION_TYPE, ""); if (partitionType.equals("PK_CHUNKING")) { workUnits = generateWorkUnitsPkChunking(sourceEntity, state, previousWatermark); } else { workUnits = generateWorkUnitsStrategy(sourceEntity, state, previousWatermark); } log.info("====Generated {} workUnit(s)====", workUnits.size()); if (workUnitConf.partitionOnly) { log.info("It is partitionOnly mode, return blank workUnit list"); return new ArrayList<>(); } else { return workUnits; } } }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { @Override protected List<WorkUnit> generateWorkUnits(SourceEntity sourceEntity, SourceState state, long previousWatermark) { List<WorkUnit> workUnits = null; workUnitConf = new SfConfig(state.getProperties()); String partitionType = state.getProp(SALESFORCE_PARTITION_TYPE, ""); if (partitionType.equals("PK_CHUNKING")) { workUnits = generateWorkUnitsPkChunking(sourceEntity, state, previousWatermark); } else { workUnits = generateWorkUnitsStrategy(sourceEntity, state, previousWatermark); } log.info("====Generated {} workUnit(s)====", workUnits.size()); if (workUnitConf.partitionOnly) { log.info("It is partitionOnly mode, return blank workUnit list"); return new ArrayList<>(); } else { return workUnits; } } SalesforceSource(); @VisibleForTesting  SalesforceSource(LineageInfo lineageInfo); }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { @Override protected List<WorkUnit> generateWorkUnits(SourceEntity sourceEntity, SourceState state, long previousWatermark) { List<WorkUnit> workUnits = null; workUnitConf = new SfConfig(state.getProperties()); String partitionType = state.getProp(SALESFORCE_PARTITION_TYPE, ""); if (partitionType.equals("PK_CHUNKING")) { workUnits = generateWorkUnitsPkChunking(sourceEntity, state, previousWatermark); } else { workUnits = generateWorkUnitsStrategy(sourceEntity, state, previousWatermark); } log.info("====Generated {} workUnit(s)====", workUnits.size()); if (workUnitConf.partitionOnly) { log.info("It is partitionOnly mode, return blank workUnit list"); return new ArrayList<>(); } else { return workUnits; } } SalesforceSource(); @VisibleForTesting  SalesforceSource(LineageInfo lineageInfo); @Override Extractor<JsonArray, JsonElement> getExtractor(WorkUnitState state); @Override boolean isEarlyStopped(); }
SalesforceSource extends QueryBasedSource<JsonArray, JsonElement> { @Override protected List<WorkUnit> generateWorkUnits(SourceEntity sourceEntity, SourceState state, long previousWatermark) { List<WorkUnit> workUnits = null; workUnitConf = new SfConfig(state.getProperties()); String partitionType = state.getProp(SALESFORCE_PARTITION_TYPE, ""); if (partitionType.equals("PK_CHUNKING")) { workUnits = generateWorkUnitsPkChunking(sourceEntity, state, previousWatermark); } else { workUnits = generateWorkUnitsStrategy(sourceEntity, state, previousWatermark); } log.info("====Generated {} workUnit(s)====", workUnits.size()); if (workUnitConf.partitionOnly) { log.info("It is partitionOnly mode, return blank workUnit list"); return new ArrayList<>(); } else { return workUnits; } } SalesforceSource(); @VisibleForTesting  SalesforceSource(LineageInfo lineageInfo); @Override Extractor<JsonArray, JsonElement> getExtractor(WorkUnitState state); @Override boolean isEarlyStopped(); static final String USE_ALL_OBJECTS; static final boolean DEFAULT_USE_ALL_OBJECTS; }