Flink CDC里请问下我有个需求怎么满足？

我有个需求，请问用Flink怎么实现？

两个flink的topic的事件流aTopic，bTopic，做join，
以aTopic作为主流，根据相近时间，值相同的某个字段进行join（比如：都有objectId和eventTime字段），并且他们上报时间和间隔都不一样。
要求aTopic的每一个记录按照相近时间join的时候，最多只能join到一个bTopic的记录，如果没有找到bTopic的记录，就把bTopic相关字段留空，如果在时间范围内，有关联到多条记录，只要最早关联到的bTopic的一条记录。

我的做法是：
1，用KafkaSource.<>builder()和env.fromSource(,WatermarkStrategy.noWatermarks(),)解析出来两个流到DataStreamSource
2,两个流分别.keyBy(objectId)，assignTimestampsAndWatermarks，withTimestampAssigner(eventTime)，设置水印。
3，tEnv.fromDataStream，columnByExpression，watermark(rowtime)，设置