@Override
    public void open() throws Exception {
        super.open();

        collector = new TimestampedCollector<>(output);
        context = new ContextImpl(userFunction);
        internalTimerService =
                getInternalTimerService(CLEANUP_TIMER_NAME, StringSerializer.INSTANCE, this);
    }
@TestTemplate
    void testReturnsCorrectTimestamp() throws Exception {
        IntervalJoinOperator<String, TestElem, TestElem, Tuple2<TestElem, TestElem>> op =
                new IntervalJoinOperator<>(
                        -1,
                        1,
                        true,
                        true,
                        null,
                        null,
                        TestElem.serializer(),
                        TestElem.serializer(),
                        new ProcessJoinFunction<TestElem, TestElem, Tuple2<TestElem, TestElem>>() {

                            private static final long serialVersionUID = 1L;

                            @Override
                            public void processElement(
                                    TestElem left,
                                    TestElem right,
                                    Context ctx,
                                    Collector<Tuple2<TestElem, TestElem>> out)
                                    throws Exception {
                                assertThat(ctx.getTimestamp())
                                        .isEqualTo(Math.max(left.ts, right.ts));
                            }
                        });

        try (TestHarness testHarness =
                new TestHarness(
                        op,
                        (elem) -> elem.key,
                        (elem) -> elem.key,
                        TypeInformation.of(String.class))) {

            testHarness.setup();
            testHarness.open();

            processElementsAndWatermarks(testHarness);
        }
    }