@Test public void testGetPartitionList() { List<Partition> expectedPartitions = new ArrayList<>(); SourceState sourceState = new SourceState(); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_IS_WATERMARK_OVERRIDE, true); TestPartitioner partitioner = new TestPartitioner(sourceState); long defaultValue = ConfigurationKeys.DEFAULT_WATERMARK_VALUE; expectedPartitions.add(new Partition(defaultValue, defaultValue, true, false)); Assert.assertEquals(partitioner.getPartitionList(-1), expectedPartitions); sourceState.setProp(ConfigurationKeys.EXTRACT_DELTA_FIELDS_KEY, "time"); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_WATERMARK_TYPE, "hour"); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_EXTRACT_TYPE, "SNAPSHOT"); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_PARTITION_INTERVAL, "2"); sourceState.setProp(ConfigurationKeys.SOURCE_MAX_NUMBER_OF_PARTITIONS, "2"); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_IS_WATERMARK_OVERRIDE, true); expectedPartitions.clear(); expectedPartitions.add(new Partition(defaultValue, Long.parseLong(TestPartitioner.currentTimeString), true, false)); Assert.assertEquals(partitioner.getPartitionList(-1), expectedPartitions); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_START_VALUE, "20170101002010"); sourceState.setProp(ConfigurationKeys.SOURCE_QUERYBASED_END_VALUE, "20170101122010"); expectedPartitions.clear(); expectedPartitions.add(new Partition(20170101000000L, 20170101060000L)); expectedPartitions.add(new Partition(20170101060000L, 20170101120000L, true, true)); List<Partition> partitions = partitioner.getPartitionList(-1); Collections.sort(partitions, Partitioner.ascendingComparator); Assert.assertEquals(partitions, expectedPartitions); }
public List<Partition> getPartitionList(long previousWatermark) { if (state.getPropAsBoolean(HAS_USER_SPECIFIED_PARTITIONS)) { return createUserSpecifiedPartitions(); } List<Partition> partitions = new ArrayList<>(); HashMap<Long, Long> partitionMap = getPartitions(previousWatermark); if (partitionMap.size() == 0) { return partitions; } if (partitionMap.size() == 1) { Map.Entry<Long, Long> entry = partitionMap.entrySet().iterator().next(); Long lwm = entry.getKey(); Long hwm = entry.getValue(); if (lwm == hwm) { if (lwm != -1) { boolean allowEqualBoundary = state.getPropAsBoolean(ALLOW_EQUAL_WATERMARK_BOUNDARY, false); LOG.info("Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary); if (!allowEqualBoundary) { return partitions; } } } } Long highestWatermark = Collections.max(partitionMap.values()); for (Map.Entry<Long, Long> entry : partitionMap.entrySet()) { Long partitionHighWatermark = entry.getValue(); if (partitionHighWatermark.equals(highestWatermark)) { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, true, hasUserSpecifiedHighWatermark)); } else { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, false)); } } return partitions; }
Partitioner { public List<Partition> getPartitionList(long previousWatermark) { if (state.getPropAsBoolean(HAS_USER_SPECIFIED_PARTITIONS)) { return createUserSpecifiedPartitions(); } List<Partition> partitions = new ArrayList<>(); HashMap<Long, Long> partitionMap = getPartitions(previousWatermark); if (partitionMap.size() == 0) { return partitions; } if (partitionMap.size() == 1) { Map.Entry<Long, Long> entry = partitionMap.entrySet().iterator().next(); Long lwm = entry.getKey(); Long hwm = entry.getValue(); if (lwm == hwm) { if (lwm != -1) { boolean allowEqualBoundary = state.getPropAsBoolean(ALLOW_EQUAL_WATERMARK_BOUNDARY, false); LOG.info("Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary); if (!allowEqualBoundary) { return partitions; } } } } Long highestWatermark = Collections.max(partitionMap.values()); for (Map.Entry<Long, Long> entry : partitionMap.entrySet()) { Long partitionHighWatermark = entry.getValue(); if (partitionHighWatermark.equals(highestWatermark)) { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, true, hasUserSpecifiedHighWatermark)); } else { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, false)); } } return partitions; } }
Partitioner { public List<Partition> getPartitionList(long previousWatermark) { if (state.getPropAsBoolean(HAS_USER_SPECIFIED_PARTITIONS)) { return createUserSpecifiedPartitions(); } List<Partition> partitions = new ArrayList<>(); HashMap<Long, Long> partitionMap = getPartitions(previousWatermark); if (partitionMap.size() == 0) { return partitions; } if (partitionMap.size() == 1) { Map.Entry<Long, Long> entry = partitionMap.entrySet().iterator().next(); Long lwm = entry.getKey(); Long hwm = entry.getValue(); if (lwm == hwm) { if (lwm != -1) { boolean allowEqualBoundary = state.getPropAsBoolean(ALLOW_EQUAL_WATERMARK_BOUNDARY, false); LOG.info("Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary); if (!allowEqualBoundary) { return partitions; } } } } Long highestWatermark = Collections.max(partitionMap.values()); for (Map.Entry<Long, Long> entry : partitionMap.entrySet()) { Long partitionHighWatermark = entry.getValue(); if (partitionHighWatermark.equals(highestWatermark)) { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, true, hasUserSpecifiedHighWatermark)); } else { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, false)); } } return partitions; } Partitioner(SourceState state); }
Partitioner { public List<Partition> getPartitionList(long previousWatermark) { if (state.getPropAsBoolean(HAS_USER_SPECIFIED_PARTITIONS)) { return createUserSpecifiedPartitions(); } List<Partition> partitions = new ArrayList<>(); HashMap<Long, Long> partitionMap = getPartitions(previousWatermark); if (partitionMap.size() == 0) { return partitions; } if (partitionMap.size() == 1) { Map.Entry<Long, Long> entry = partitionMap.entrySet().iterator().next(); Long lwm = entry.getKey(); Long hwm = entry.getValue(); if (lwm == hwm) { if (lwm != -1) { boolean allowEqualBoundary = state.getPropAsBoolean(ALLOW_EQUAL_WATERMARK_BOUNDARY, false); LOG.info("Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary); if (!allowEqualBoundary) { return partitions; } } } } Long highestWatermark = Collections.max(partitionMap.values()); for (Map.Entry<Long, Long> entry : partitionMap.entrySet()) { Long partitionHighWatermark = entry.getValue(); if (partitionHighWatermark.equals(highestWatermark)) { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, true, hasUserSpecifiedHighWatermark)); } else { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, false)); } } return partitions; } Partitioner(SourceState state); Partition getGlobalPartition(long previousWatermark); @Deprecated HashMap<Long, Long> getPartitions(long previousWatermark); List<Partition> getPartitionList(long previousWatermark); boolean isFullDump(); boolean isWatermarkOverride(); @VisibleForTesting DateTime getCurrentTime(String timeZone); }
Partitioner { public List<Partition> getPartitionList(long previousWatermark) { if (state.getPropAsBoolean(HAS_USER_SPECIFIED_PARTITIONS)) { return createUserSpecifiedPartitions(); } List<Partition> partitions = new ArrayList<>(); HashMap<Long, Long> partitionMap = getPartitions(previousWatermark); if (partitionMap.size() == 0) { return partitions; } if (partitionMap.size() == 1) { Map.Entry<Long, Long> entry = partitionMap.entrySet().iterator().next(); Long lwm = entry.getKey(); Long hwm = entry.getValue(); if (lwm == hwm) { if (lwm != -1) { boolean allowEqualBoundary = state.getPropAsBoolean(ALLOW_EQUAL_WATERMARK_BOUNDARY, false); LOG.info("Single partition with LWM = HWM and allowEqualBoundary=" + allowEqualBoundary); if (!allowEqualBoundary) { return partitions; } } } } Long highestWatermark = Collections.max(partitionMap.values()); for (Map.Entry<Long, Long> entry : partitionMap.entrySet()) { Long partitionHighWatermark = entry.getValue(); if (partitionHighWatermark.equals(highestWatermark)) { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, true, hasUserSpecifiedHighWatermark)); } else { partitions.add(new Partition(entry.getKey(), partitionHighWatermark, false)); } } return partitions; } Partitioner(SourceState state); Partition getGlobalPartition(long previousWatermark); @Deprecated HashMap<Long, Long> getPartitions(long previousWatermark); List<Partition> getPartitionList(long previousWatermark); boolean isFullDump(); boolean isWatermarkOverride(); @VisibleForTesting DateTime getCurrentTime(String timeZone); static final String WATERMARKTIMEFORMAT; static final String HAS_USER_SPECIFIED_PARTITIONS; static final String USER_SPECIFIED_PARTITIONS; static final String IS_EARLY_STOPPED; static final String ALLOW_EQUAL_WATERMARK_BOUNDARY; static final Comparator<Partition> ascendingComparator; }