У меня есть топи c, который содержит события подключения и отключения пользователя для каждой сессии. Я хотел бы использовать поток Kafka для обработки этого topi c и обновления KTable в зависимости от некоторых условий. Каждая запись не может обновлять KTable. Поэтому мне нужно обработать несколько записей, чтобы узнать, нужно ли обновлять KTable.
Например, поток обработки и агрегирование по пользователю, а затем по идентификатору сеанса. Если хотя бы один sessionid этого пользователя имеет только событие Connected, KTable должен быть обновлен как пользователь в сети, если еще не обновлен.
Если все sessionId пользователя имеют событие Disconnected, KTable должен быть обновлен как пользователь в автономном режиме, если еще не обновлен.
Как я могу реализовать такой лог? c?
Можем ли мы реализовать эту KTable во всех экземплярах приложения, чтобы каждый экземпляр имел эти данные локально?