@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
    public void testWindowFunctionWithAtmostOnce() throws Exception {
        windowConfig.setProcessingGuarantees(WindowConfig.ProcessingGuarantees.ATMOST_ONCE);
        doReturn(Optional.of(new Gson().fromJson(new Gson().toJson(windowConfig), Map.class))).when(context)
                .getUserConfigValue(WindowConfig.WINDOW_CONFIG_KEY);
        Record record = mock(Record.class);
        when(context.getCurrentRecord()).thenReturn(record);
        doReturn(Optional.of("test-topic")).when(record).getTopicName();
        doReturn(record).when(context).getCurrentRecord();
        doReturn(100l).when(record).getValue();
        testWindowedPulsarFunction.process(10L, context);
        verify(record, times(1)).ack();
    }