Для выполнения соединения в Kafka Streams необходимо, чтобы все его участники были совместно секционированы, то есть у них было одинаковое число секций и ключи одного типа. В результате при вызове метода join() в листинге 4.13 оба экземпляра KStream проверяются на предмет необходимости повторного секционирования.
Примечание
Участвующие в соединении экземпляры интерфейса GlobalKTable повторного секционирования не требуют.
В подразделе 4.4.2 мы применяли к transactionStream метод selectKey() и сразу после этого осуществляли ветвление по возвращаемым объектам KStream. А поскольку метод selectKey() изменяет ключи, то как объект coffeeStream, так и electronicsStream требуют повторного секционирования. Имеет смысл еще раз сказать, что повторное секционирование требуется для того, чтобы гарантировать попадание записей с идентичными ключами в одну секцию. Повторное секционирование выполняется автоматически. Кроме того, при запуске приложения Kafka Streams проверяется, одинаковое ли число секций в участвующих в соединении топиках, и при обнаружении расхождений генерирует исключение TopologyBuilderException. За то, чтобы участвующие в соединении ключи были одного типа, отвечает разработчик.
Совместное секционирование также требует, чтобы все генераторы Kafka использовали один и тот же класс секционирования при записи в топики-источники Kafka Streams. Аналогично вы должны применять один и тот же объект StreamPartitioner при любых операциях записи в топики-стоки Kafka Streams посредством метода KStream.to(). Если же вы придерживаетесь стратегий секционирования по умолчанию, то можете о них не волноваться.
Продолжим разговор о соединениях и посмотрим на остальные их варианты.
Кіру не тіркелу пікір қалдыру үшін