запрос на левое соединение без результатов

Question

У меня проблема с левым соединением, похоже, не работает. Для внутреннего присоединения это нормально. Я использую поточный коннектор Pulsar-Spark.

Dataset datasetPerson = ...
Dataset datasetPersonWithWt = datasetPerson.withWatermark("__eventTime", "20 seconds").as("per");
Dataset datasetPhone = ...
Dataset datasetPhoneWithWt = datasetPolicy.withWatermark("__eventTime", "20 seconds").as("ph");

Dataset joined = datasetPersonWithWt.join(
   datasetPhoneWithWt,
   // functions.expr("per.__key=ph.__key"),
   functions.expr("per.__key=ph.__key and ph.__eventTime >= per.__eventTime - interval 20 seconds and ph.__eventTime <= per.__eventTime + interval 20 seconds"),
"left");

Что может быть не так с запросом?

Еще интересно, коннектор подключается к пульсару в ЭКСКЛЮЗИВНОМ режиме. Гарантируется ли Spark, что все сообщения с одним и тем же ключом обрабатываются одним и тем же воркером? Мне нужно собрать полный объект и сохранить его в db