@Test public void testExecuteWithLateTupleStream() throws Exception { windowConfig.setLateDataTopic("$late"); doReturn(Optional.of(new Gson().fromJson(new Gson().toJson(windowConfig), Map.class))) .when(context).getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY); TypedMessageBuilder typedMessageBuilder = mock(TypedMessageBuilder.class); when(typedMessageBuilder.value(any())).thenReturn(typedMessageBuilder); when(typedMessageBuilder.sendAsync()).thenReturn(CompletableFuture.anyOf()); when(context.newOutputMessage(anyString(), any())).thenReturn(typedMessageBuilder); long[] timestamps = {603, 605, 607, 618, 626, 636, 600}; List<Long> events = new ArrayList<>(timestamps.length); for (long ts : timestamps) { events.add(ts); Record<?> record = mock(Record.class); doReturn(Optional.of("test-topic")).when(record).getTopicName(); doReturn(record).when(context).getCurrentRecord(); doReturn(ts).when(record).getValue(); testWindowedPulsarFunction.process(ts, context); testWindowedPulsarFunction.waterMarkEventGenerator.run(); } System.out.println(testWindowedPulsarFunction.windows); long event = events.get(events.size() - 1); }
@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);  }