anchor.diff:  ----
   
   The pipeline logic itself is defined in https://github.com/debezium/debezium-examples/blob/master/camel-component/qa-camel/src/main/java/io/debezium/examples/camel/pipeline/QaDatabaseUserNotifier.java[QaDatabaseUserNotifier] class.
  -
   Its main route looks like tis:
   
   [source,java]
  
positive.code:               .log(LoggingLevel.TRACE, "Converted to logical class ${body}")
                .bean(store, "readFromStoreAndAddAnswer")
                .to(ROUTE_MAIL_ANSWER_CHANGE)
                .filter(hasManyAnswers)                                            // (7)
                  .setBody().simple("Question '${exchangeProperty[aggregate].text}' has many answers (generated at " + Instant.now() + ")")
                  .to(TWITTER_SERVER)
                .end()
              .endChoice()
            .otherwise()
              .log(LoggingLevel.WARN, "Unknown type ${headers[" + DebeziumConstants.HEADER_IDENTIFIER + "]}")
          .endParent();
  
      from(ROUTE_MAIL_QUESTION_CREATE)                                             // (6)
        .routeId(QaDatabaseUserNotifier.class.getName() + ".QuestionNotifier")
        .setHeader("To").simple("${body.email}")
        .setHeader("Subject").simple("Question created/edited")
        .setBody().simple("Question '${body.text}' was created or edited")
        .to(SMTP_SERVER);
    }
  }
  
  @Converter
  public class Converters {
  
    @Converter
    public static Question questionFromStruct(Struct struct) {                     // (4)
      return new Question(struct.getInt64("id"), struct.getString("text"), struct.getString("email"));
    }
  }
  ----
  
  1. `from` is the Debezium source endpoint. The URI parts map directly to connector configuration options.
  2. The pipeline logic is split depending on the change event type.
  The recognition is based on `CamelDebeziumIdentifier` header which contains the identifier (`<server_name>.<schema_name>.<table_name>`) of the source table.
  3. The pipeline is now able to process only updates and deletes.
  The recognition is based on `CamelDebeziumOperation` header that contains `op` field of the message `Envelope`.
  4. The Kafka Connect's `Struct` type is converted into a logical type used in the pipeline.
  The conversion is performed by a custom Camel converter.
  It is possible to use out-of-the-box `DebeziumTypeConverter` that converts `Struct` into a `Map` but this tightly couples pipeline logic into the table structure.
  5. A supplementary route is invoked that communicates with a https://github.com/debezium/debezium-examples/blob/master/camel-component/qa-camel/src/main/java/io/debezium/examples/camel/pipeline/AggregateStore.java[message store] based on an Infinispan cache to build a message aggregate.
  The message store checks if it has the question already stored.
  If not a new aggregate is created and stored otherwise the stored aggregate is updated with new data.
  6. A supplementary route is invoked that formats a mail message and delivers it to the question creator via the SMTP endpoint.
  7. The route part related to the answer message type is very similar (answers are added to question aggregate).
  The main difference is the posting of a Twitter message when the aggregate contains three answers.
  
  == Demo
  
negative.code: *Figure 1. The Deployment Topology*
  ====
  
  * The Q & A application is implemented using the https://quarkus.io/[Quarkus] stack and exposes a REST API to create questions and answers
  * The application stores its data in the PostgreSQL database
  * The Camel route runs as a plain Java application that uses an embedded https://infinispan.org/[Infinispan] store to persist its state (used to build an aggregate object linking questions to their answers) and sends messages to about answered questions via e-mail and to an associated Twitter account
  * A https://github.com/mailhog/MailHog[MailHog] SMTP server running in a container for sending e-mails
  
  === Question & Answer Application
  
  