anchor.diff:          + "&schemaWhitelist={{database.schema}}"
           + "&tableWhitelist={{database.schema}}.question,{{database.schema}}.answer"
           + "&offsetStorage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore")
  -        .routeId(QaDatabaseUserNotifier.class.getName() + ".DatabaseReader")     // (1)
  +        .routeId(QaDatabaseUserNotifier.class.getName() + ".DatabaseReader")     // <1>
           .log(LoggingLevel.DEBUG, "Incoming message ${body} with headers ${headers}")
  -        .choice()                                                                // (2)
  +        .choice()                                                                // <2>
             .when(isQuestionEvent)
  -            .filter(isCreateOrUpdateEvent)                                       // (3)
  -              .convertBodyTo(Question.class)                                     // (4)
  +            .filter(isCreateOrUpdateEvent)                                       // <3>
  +              .convertBodyTo(Question.class)                                     // <4>
                 .log(LoggingLevel.TRACE, "Converted to logical class ${body}")
  -              .bean(store, "readFromStoreAndUpdateIfNeeded")                     // (5)
  -              .to(ROUTE_MAIL_QUESTION_CREATE)                                    // (6)
  +              .bean(store, "readFromStoreAndUpdateIfNeeded")                     // <5>
  +              .to(ROUTE_MAIL_QUESTION_CREATE)                                    // <6>
               .endChoice()
             .when(isAnswerEvent)
               .filter(isCreateOrUpdateEvent)
  
positive.code: ----
  
  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]
  
negative.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 static class Converters {
  
      @Converter
      public static Question questionFromStruct(Struct struct) {                   // <4>
        return new Question(struct.getInt64("id"), struct.getString("text"),
            struct.getString("email"));
      }
  
      @Converter
      public static Answer answerFromStruct(Struct struct) {                       // <4>
        return new Answer(struct.getInt64("id"), struct.getString("text"),
            struct.getString("email"), struct.getInt64("question_id"));
      }
    }
  }
  ----
  <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
  