Spring Cloud Stream: BiConsumer

application.yml

spring:
  cloud:
    stream:
      bindings:
        processPersonEvents-in-0:
          destination: person-changed-events
        processPersonEvents-in-1:
          destination: address-changed-events
      default:
        resetOffsets: true
        startOffsets: earliest
      kafka:
        streams:
          binder:
            configuration:
              num.stream.threads: 2
...

BiConsumerConfig.java

@Bean
public BiConsumer<KTable<String, PersonChangedEvent>, KTable<String, AddressChangedEvent>> processPersonEvents(
        ProjectorService projectorService,
        MappingContext mappingContext) {
 
  return (personChangedEvents, addressChangedEvents) -> {
     
    personChangedEvents
            .mapValues(personChangedEvent -> PersonAggregateRoot.empty().apply(personChangedEvent, mappingContext))
            .join(addressChangedEvents, ((root, addressChangedEvent) -> root.apply(addressChangedEvent, mappingContext)))
            .mapValues(projectorService::project)
            .mapValues(PersonAggregateRoot::getPersonId)
            .toStream()
            .foreach((k, v) -> log.info("Read model updated for key '{}' and personId '{}'", k, v));
  };
}