@Override
    public void run() {
        // initialize the thread context
        ThreadContext.put("function", WindowUtils.getFullyQualifiedName(
                context.getTenant(), context.getNamespace(), context.getFunctionName()));
        try {
            long waterMarkTs = computeWaterMarkTs();
            if (waterMarkTs > lastWaterMarkTs) {
                windowManager.add(new WaterMarkEvent<>(waterMarkTs));
                lastWaterMarkTs = waterMarkTs;
            }
        } catch (Throwable th) {
            log.error("Failed while processing watermark event ", th);
            throw th;
        }
    }
@Test
    public void testNoEvents() throws Exception {
        waterMarkEventGenerator.run();
        assertTrue(eventList.isEmpty());
    }