@Test(expectedExceptions = RuntimeException.class) public void testExecuteWithWrongWrongTimestampExtractorType() throws Exception { WindowConfig windowConfig = new WindowConfig(); windowConfig.setTimestampExtractorClassName(TestWrongTimestampExtractor.class.getName()); Mockito.doReturn(Optional.of(new Gson().fromJson(new Gson().toJson(windowConfig), Map.class))) .when(context).getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY); testWindowedPulsarFunction.process(10L, context); }
@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);  }