@Test public static void testSweep() { Random random = new Random(); for (int j =0; j < 1000; ++j) { FineGrainedWatermarkTracker tracker = new FineGrainedWatermarkTracker(ConfigFactory.empty()); tracker.setAutoStart(false); int numWatermarks = 1+random.nextInt(1000); AcknowledgableWatermark[] acknowledgableWatermarks = new AcknowledgableWatermark[numWatermarks]; for (int i = 0; i < numWatermarks; ++i) { CheckpointableWatermark checkpointableWatermark = new DefaultCheckpointableWatermark("default", new LongWatermark(i)); AcknowledgableWatermark ackable = new AcknowledgableWatermark(checkpointableWatermark); acknowledgableWatermarks[i] = ackable; tracker.track(ackable); } int numMissingAcks = random.nextInt(numWatermarks); SortedSet<Integer> holes = new TreeSet<>(); for (int i = 0; i < numMissingAcks; ++i) { holes.add(random.nextInt(numWatermarks)); } for (int i = 0; i < numWatermarks; ++i) { if (!holes.contains(i)) { acknowledgableWatermarks[i].ack(); } } verifyCommitables(tracker, holes, numWatermarks - 1); int swept = tracker.sweep(); if (holes.isEmpty()) { Assert.assertEquals(swept, numWatermarks -1); } else { if (holes.contains(0)) { Assert.assertEquals(swept, 0); } else { Assert.assertEquals(swept, holes.first() - 1); } } verifyCommitables(tracker, holes, numWatermarks - 1); } }
@VisibleForTesting synchronized int sweep() { long startTime = System.nanoTime(); int swept = 0; for (Map.Entry<String, Deque<AcknowledgableWatermark>> entry : _watermarksMap.entrySet()) { Deque<AcknowledgableWatermark> watermarks = entry.getValue(); boolean continueIteration = true; while (continueIteration) { Iterator<AcknowledgableWatermark> iter = watermarks.iterator(); if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark first = iter.next(); if (first.isAcked()) { if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark second = iter.next(); if ((second != null) && second.isAcked()) { watermarks.pop(); swept++; } else { continueIteration = false; } } else { continueIteration = false; } } } long duration = (System.nanoTime() - startTime)/ MILLIS_TO_NANOS; log.debug("Swept {} watermarks in {} millis", swept, duration); _watermarksSwept.mark(swept); return swept; }
FineGrainedWatermarkTracker implements Instrumentable, Closeable { @VisibleForTesting synchronized int sweep() { long startTime = System.nanoTime(); int swept = 0; for (Map.Entry<String, Deque<AcknowledgableWatermark>> entry : _watermarksMap.entrySet()) { Deque<AcknowledgableWatermark> watermarks = entry.getValue(); boolean continueIteration = true; while (continueIteration) { Iterator<AcknowledgableWatermark> iter = watermarks.iterator(); if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark first = iter.next(); if (first.isAcked()) { if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark second = iter.next(); if ((second != null) && second.isAcked()) { watermarks.pop(); swept++; } else { continueIteration = false; } } else { continueIteration = false; } } } long duration = (System.nanoTime() - startTime)/ MILLIS_TO_NANOS; log.debug("Swept {} watermarks in {} millis", swept, duration); _watermarksSwept.mark(swept); return swept; } }
FineGrainedWatermarkTracker implements Instrumentable, Closeable { @VisibleForTesting synchronized int sweep() { long startTime = System.nanoTime(); int swept = 0; for (Map.Entry<String, Deque<AcknowledgableWatermark>> entry : _watermarksMap.entrySet()) { Deque<AcknowledgableWatermark> watermarks = entry.getValue(); boolean continueIteration = true; while (continueIteration) { Iterator<AcknowledgableWatermark> iter = watermarks.iterator(); if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark first = iter.next(); if (first.isAcked()) { if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark second = iter.next(); if ((second != null) && second.isAcked()) { watermarks.pop(); swept++; } else { continueIteration = false; } } else { continueIteration = false; } } } long duration = (System.nanoTime() - startTime)/ MILLIS_TO_NANOS; log.debug("Swept {} watermarks in {} millis", swept, duration); _watermarksSwept.mark(swept); return swept; } FineGrainedWatermarkTracker(Config config); }
FineGrainedWatermarkTracker implements Instrumentable, Closeable { @VisibleForTesting synchronized int sweep() { long startTime = System.nanoTime(); int swept = 0; for (Map.Entry<String, Deque<AcknowledgableWatermark>> entry : _watermarksMap.entrySet()) { Deque<AcknowledgableWatermark> watermarks = entry.getValue(); boolean continueIteration = true; while (continueIteration) { Iterator<AcknowledgableWatermark> iter = watermarks.iterator(); if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark first = iter.next(); if (first.isAcked()) { if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark second = iter.next(); if ((second != null) && second.isAcked()) { watermarks.pop(); swept++; } else { continueIteration = false; } } else { continueIteration = false; } } } long duration = (System.nanoTime() - startTime)/ MILLIS_TO_NANOS; log.debug("Swept {} watermarks in {} millis", swept, duration); _watermarksSwept.mark(swept); return swept; } FineGrainedWatermarkTracker(Config config); void track(AcknowledgableWatermark acknowledgableWatermark); Map<String, CheckpointableWatermark> getCommittableWatermarks(); Map<String, CheckpointableWatermark> getUnacknowledgedWatermarks(); synchronized void start(); @Override void close(); @Override void switchMetricContext(List<Tag<?>> tags); @Override void switchMetricContext(MetricContext context); @Override List<Tag<?>> generateTags(State state); @Nonnull @Override MetricContext getMetricContext(); @Override boolean isInstrumentationEnabled(); }
FineGrainedWatermarkTracker implements Instrumentable, Closeable { @VisibleForTesting synchronized int sweep() { long startTime = System.nanoTime(); int swept = 0; for (Map.Entry<String, Deque<AcknowledgableWatermark>> entry : _watermarksMap.entrySet()) { Deque<AcknowledgableWatermark> watermarks = entry.getValue(); boolean continueIteration = true; while (continueIteration) { Iterator<AcknowledgableWatermark> iter = watermarks.iterator(); if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark first = iter.next(); if (first.isAcked()) { if (!iter.hasNext()) { continueIteration = false; continue; } AcknowledgableWatermark second = iter.next(); if ((second != null) && second.isAcked()) { watermarks.pop(); swept++; } else { continueIteration = false; } } else { continueIteration = false; } } } long duration = (System.nanoTime() - startTime)/ MILLIS_TO_NANOS; log.debug("Swept {} watermarks in {} millis", swept, duration); _watermarksSwept.mark(swept); return swept; } FineGrainedWatermarkTracker(Config config); void track(AcknowledgableWatermark acknowledgableWatermark); Map<String, CheckpointableWatermark> getCommittableWatermarks(); Map<String, CheckpointableWatermark> getUnacknowledgedWatermarks(); synchronized void start(); @Override void close(); @Override void switchMetricContext(List<Tag<?>> tags); @Override void switchMetricContext(MetricContext context); @Override List<Tag<?>> generateTags(State state); @Nonnull @Override MetricContext getMetricContext(); @Override boolean isInstrumentationEnabled(); static final String WATERMARK_TRACKER_SWEEP_INTERVAL_MS; static final Long WATERMARK_TRACKER_SWEEP_INTERVAL_MS_DEFAULT; }