@Test public void testExecuteWithTs() throws Exception { long[] timestamps = {603, 605, 607, 618, 626, 636}; for (long ts : timestamps) { Record<?> record = Mockito.mock(Record.class); Mockito.doReturn(Optional.of("test-topic")).when(record).getTopicName(); Mockito.doReturn(record).when(context).getCurrentRecord(); Mockito.doReturn(ts).when(record).getValue(); testWindowedPulsarFunction.process(ts, context); } testWindowedPulsarFunction.waterMarkEventGenerator.run(); assertEquals(3, testWindowedPulsarFunction.windows.size()); Window<Record<Long>> first = testWindowedPulsarFunction.windows.get(0); assertArrayEquals( new long[]{603, 605, 607}, new long[]{first.get().get(0).getValue(), first.get().get(1).getValue(), first.get().get(2).getValue()}); Window<Record<Long>> second = testWindowedPulsarFunction.windows.get(1); assertArrayEquals( new long[]{603, 605, 607, 618}, new long[]{second.get().get(0).getValue(), second.get().get(1).getValue(), second.get().get(2).getValue(), second.get().get(3).getValue()}); Window<Record<Long>> third = testWindowedPulsarFunction.windows.get(2); assertArrayEquals(new long[]{618, 626}, new long[]{third.get().get(0).getValue(), third.get().get(1).getValue()}); }
@Override public O process(I input, Context context) throws Exception { if (!this.initialized) { initialize(context); } Record<I> record = (Record<I>)context.getCurrentRecord(); if (isEventTime()) { long ts = this.timestampExtractor.extractTimestamp(record.getValue()); if (this.waterMarkEventGenerator.track(record.getTopicName().get(), ts)) { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" + ".", input, ts)); } record.ack(); } } else { this.windowManager.add(record, System.currentTimeMillis(), record); } return null; }
WindowFunctionExecutor implements Function<I, O> { @Override public O process(I input, Context context) throws Exception { if (!this.initialized) { initialize(context); } Record<I> record = (Record<I>)context.getCurrentRecord(); if (isEventTime()) { long ts = this.timestampExtractor.extractTimestamp(record.getValue()); if (this.waterMarkEventGenerator.track(record.getTopicName().get(), ts)) { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" + ".", input, ts)); } record.ack(); } } else { this.windowManager.add(record, System.currentTimeMillis(), record); } return null; } }
WindowFunctionExecutor implements Function<I, O> { @Override public O process(I input, Context context) throws Exception { if (!this.initialized) { initialize(context); } Record<I> record = (Record<I>)context.getCurrentRecord(); if (isEventTime()) { long ts = this.timestampExtractor.extractTimestamp(record.getValue()); if (this.waterMarkEventGenerator.track(record.getTopicName().get(), ts)) { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" + ".", input, ts)); } record.ack(); } } else { this.windowManager.add(record, System.currentTimeMillis(), record); } return null; }  }
WindowFunctionExecutor implements Function<I, O> { @Override public O process(I input, Context context) throws Exception { if (!this.initialized) { initialize(context); } Record<I> record = (Record<I>)context.getCurrentRecord(); if (isEventTime()) { long ts = this.timestampExtractor.extractTimestamp(record.getValue()); if (this.waterMarkEventGenerator.track(record.getTopicName().get(), ts)) { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" + ".", input, ts)); } record.ack(); } } else { this.windowManager.add(record, System.currentTimeMillis(), record); } return null; }  void initialize(Context context); void shutdown(); @Override O process(I input, Context context); O process(Window<Record<I>> inputWindow, WindowContext context); }
WindowFunctionExecutor implements Function<I, O> { @Override public O process(I input, Context context) throws Exception { if (!this.initialized) { initialize(context); } Record<I> record = (Record<I>)context.getCurrentRecord(); if (isEventTime()) { long ts = this.timestampExtractor.extractTimestamp(record.getValue()); if (this.waterMarkEventGenerator.track(record.getTopicName().get(), ts)) { this.windowManager.add(record, ts, record); } else { if (this.windowConfig.getLateDataTopic() != null) { context.newOutputMessage(this.windowConfig.getLateDataTopic(), null).value(input).sendAsync(); } else { log.info(String.format( "Received a late tuple %s with ts %d. This will not be " + "processed" + ".", input, ts)); } record.ack(); } } else { this.windowManager.add(record, System.currentTimeMillis(), record); } return null; }  void initialize(Context context); void shutdown(); @Override O process(I input, Context context); O process(Window<Record<I>> inputWindow, WindowContext context);  }