registerEventTimeTimer is not firing with watermark

I'm using KafkaSource to read kafka messages of Events type, as per documentation providing event time extractor is optional for source kafka

  KafkaSource<Events> source =
                KafkaSource.<Events>builder()
                        .setProperties(kafkaProperties)
                        .setBootstrapServers(parameters.get("bootstrap-servers-source"))
                        .setTopics(parameters.get("source-topic"))
                        .setGroupId("visit-events-flink-mvp")
                        .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                        //.setStartingOffsets(OffsetsInitializer.earliest())
                        .setValueOnlyDeserializer(new EventsDeserializationSchema())
                        .build();

        // event stream from kafka source
        DataStream<Events> eventStream =
                env.fromSource(source, WatermarkStrategy.forMonotonousTimestamps(), "Kafka Source")
                        //should be a unique id
                        .uid("kafka-source");

        //stream is keyed based on the anonymousId
        DataStream<Events> keyedStream =
                eventStream.keyBy(Events::getAnonymousId)
                        //  .process(new KeyedProcessing(Long.parseLong(parameters.get("ttl"))))
                        .process(new KeyedProcessingWithCallBack(Long.parseLong(parameters.get("ttl"))))
                        .uid("engager-events-keyed-processing");

In my KeyedProcessingWithCallBack, I'm setting event time timer for 60 secs and the call back is not triggering at all.

My kafka source has 8 partitions and I'm running job with parallelism 1

 public void processElement(EngagerEvents value, KeyedProcessFunction<String, EngagerEvents, String>.Context ctx, Collector<String> out) throws Exception {
        ObjectMapper objectMapper = new ObjectMapper();
        JsonNode jsonNode = objectMapper.readTree(value.getEventString());
        System.out.println("time : " +jsonNode.get("EVENT_TIMESTAMP").textValue());
        if (anonymousIdHasBeenSeen.value() == null) {
            System.out.println("time stamp emitting: " +jsonNode.get("EVENT_TIMESTAMP").textValue());
            // key is not available in the state
            anonymousIdHasBeenSeen.update(true);
            System.out.println("TIMER START TIME: " +ctx.timestamp());
            out.collect(value.getEventString());
            ctx.timerService().registerEventTimeTimer(ctx.timestamp() + (stateTtl * 1000));
        }
    }

    

    // not getting triggered
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out)
            throws Exception {
        // triggers after ttl has passed
        System.out.println("Call back triggered : time : " +timestamp + " value : " +anonymousIdHasBeenSeen.value());
       anonymousIdHasBeenSeen.clear();
    }

TEST Simulator code which will send event with anonymousId=111 with different event time stamp

try {
            for (int i = 0; i < 500; i++) {
                String[] anonymousId = {"111"};
                String key = String.valueOf(new Random().nextInt(10));
                ProducerRecord<String, String> record = new ProducerRecord<>(
                        "flink-visits-mvp-test-source",
                        key,
                        //   getEvent(UUID.randomUUID().toString() + "-"  +Thread.currentThread().getName() , event[new Random().nextInt(1)]));
                        // getEvent(anonymousId[new Random().nextInt(1)], event[new Random().nextInt(1)]));
                        getEvent(anonymousId[new Random().nextInt(1)],
                                System.currentTimeMillis(),
                                event));

                //System.out.println(record.value().toString());
                    producer.send(record);
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }

Am I doing something wrong here ? Why is my event time timer call back not trgiggering ?