@Test(expectedExceptions = IllegalStateException.class) public void testMoreThanOneWatermarkWorkunits() throws Exception { WorkUnitState previousWus = new WorkUnitState(); previousWus.setProp(ConfigurationKeys.DATASET_URN_KEY, "test_dataset_urn"); previousWus.setProp(PartitionLevelWatermarker.IS_WATERMARK_WORKUNIT_KEY, true); previousWus.setActualHighWatermark(new MultiKeyValueLongWatermark(ImmutableMap.of("2015", 100l))); WorkUnitState previousWus2 = new WorkUnitState(); previousWus2.setProp(ConfigurationKeys.DATASET_URN_KEY, "test_dataset_urn"); previousWus2.setProp(PartitionLevelWatermarker.IS_WATERMARK_WORKUNIT_KEY, true); previousWus2.setActualHighWatermark(new MultiKeyValueLongWatermark(ImmutableMap.of("2016", 101l))); SourceState state = new SourceState(new State(), Lists.newArrayList(previousWus, previousWus2)); new PartitionLevelWatermarker(state); }
@Override public void setActualHighWatermark(WorkUnitState wus) { if (Boolean.valueOf(wus.getPropAsBoolean(IS_WATERMARK_WORKUNIT_KEY))) { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(MultiKeyValueLongWatermark.class)); } else { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(LongWatermark.class)); } }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void setActualHighWatermark(WorkUnitState wus) { if (Boolean.valueOf(wus.getPropAsBoolean(IS_WATERMARK_WORKUNIT_KEY))) { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(MultiKeyValueLongWatermark.class)); } else { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(LongWatermark.class)); } } }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void setActualHighWatermark(WorkUnitState wus) { if (Boolean.valueOf(wus.getPropAsBoolean(IS_WATERMARK_WORKUNIT_KEY))) { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(MultiKeyValueLongWatermark.class)); } else { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(LongWatermark.class)); } } PartitionLevelWatermarker(State state); }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void setActualHighWatermark(WorkUnitState wus) { if (Boolean.valueOf(wus.getPropAsBoolean(IS_WATERMARK_WORKUNIT_KEY))) { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(MultiKeyValueLongWatermark.class)); } else { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(LongWatermark.class)); } } PartitionLevelWatermarker(State state); @Override void onTableProcessBegin(Table table, long tableProcessTime); @Override void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime); @Override LongWatermark getPreviousHighWatermark(Table table); @Override LongWatermark getPreviousHighWatermark(Partition partition); @Override void onGetWorkunitsEnd(List<WorkUnit> workunits); @Override LongWatermark getExpectedHighWatermark(Table table, long tableProcessTime); @Override LongWatermark getExpectedHighWatermark(Partition partition, long tableProcessTime, long partitionProcessTime); @Override void setActualHighWatermark(WorkUnitState wus); @VisibleForTesting static String tableKey(Table table); @VisibleForTesting static String partitionKey(Partition partition); }
PartitionLevelWatermarker implements HiveSourceWatermarker { @Override public void setActualHighWatermark(WorkUnitState wus) { if (Boolean.valueOf(wus.getPropAsBoolean(IS_WATERMARK_WORKUNIT_KEY))) { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(MultiKeyValueLongWatermark.class)); } else { wus.setActualHighWatermark(wus.getWorkunit().getExpectedHighWatermark(LongWatermark.class)); } } PartitionLevelWatermarker(State state); @Override void onTableProcessBegin(Table table, long tableProcessTime); @Override void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime); @Override LongWatermark getPreviousHighWatermark(Table table); @Override LongWatermark getPreviousHighWatermark(Partition partition); @Override void onGetWorkunitsEnd(List<WorkUnit> workunits); @Override LongWatermark getExpectedHighWatermark(Table table, long tableProcessTime); @Override LongWatermark getExpectedHighWatermark(Partition partition, long tableProcessTime, long partitionProcessTime); @Override void setActualHighWatermark(WorkUnitState wus); @VisibleForTesting static String tableKey(Table table); @VisibleForTesting static String partitionKey(Partition partition); static final String IS_WATERMARK_WORKUNIT_KEY; }