@Override
    public X process(T input, Context context) throws Exception {
        if (!this.initialized) {
            initialize(context);
        }

        // record must be PulsarFunctionRecord.
        Record<T> record = (Record<T>) context.getCurrentRecord();

        // windows function processing semantics requires separate processing
        if (windowConfig.getProcessingGuarantees() == WindowConfig.ProcessingGuarantees.ATMOST_ONCE) {
            record.ack();
        }

        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));
                }
            }
        } else {
            this.windowManager.add(record, System.currentTimeMillis(), record);
        }
        return null;
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void testExecuteWithWrongJavaWindowFunctionType() throws Exception {
        WindowConfig windowConfig = new WindowConfig();
        windowConfig.setActualWindowFunctionClassName(TestWrongFunction.class.getName());
        doReturn(Optional.of(new Gson().fromJson(new Gson().toJson(windowConfig), Map.class)))
                .when(context).getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY);

        testWindowedPulsarFunction.process(10L, context);
    }