anchor.diff:  `StreamsBuilder#table()` is used to read the two Kafka topics into the KTable `addresses` and `customers`, respectively
   <5> The message from the `addresses` topic is joined with the corresponding `customers` topic; the join result contains the data of the customer with one of their addresses
   <6> `groupBy()` operation will have the records to be grouped by `customer#id`
  -<7> To produce the nested structure of one customer and all their addresses, the `aggregate()` operation is applied to each group of records (customer-address tuple), updating an `CustomerWithAddresses` per customer
  +<7> To produce the nested structure of one customer and all their addresses, the `aggregate()` operation is applied to each group of records (customer-address tuples), updating a `CustomerWithAddresses` per customer
   <8> The results of the pipeline are written out to the `customersWithAddressesTopic` topic
   
   The `CustomerWithAddresses` class keeps track of the aggregated values while the events are processed in the streaming pipeline.
  
positive.code: As stated above, by default Debezium will emit events for each table on distinct topics.
  Using Kafka Streams, the change event topics for both tables will be loaded into two ``KTable``s, which are joined on the customer id.
  The Kafka Streams application is going to process data from the two Kafka topics.
  Whenever there's a new CDC event on either topic -- triggered by the insertion, update, or deletion of a record -- the join will be re-triggered.
  
  As a runtime for the Kafka Streams application, we're going to use https://quarkus.io/[Quarkus], a stack for building cloud-native microservices, which (amongst many others) also provides an https://quarkus.io/guides/kafka-streams[extension] for Kafka Streams. While it's general possible to run a Kafka Streams topology via a plain `main()` method, using Quarkus and this extension as a foundation has a number of advantages:
  
  
negative.code:         debezium/tooling:1.1 \
          bash -c 'pgcli postgresql://postgres:postgres@postgres:5432/postgres'
  
  # in pgcli, e.g. to update a customer record:
  
  > update inventory.customers set first_name = 'Sarah' where id = 1001;
  ----
  