anchor.diff:  
   [source,java]
   ----
  +public class QaDatabaseUserNotifier extends RouteBuilder {
  +
  +  @Override
  +  public void configure() throws Exception {
       from("debezium-postgres:localhost?"
  -            + "databaseHostname={{database.hostname}}"
  -            + "&databasePort={{database.port}}"
  -            + "&databaseUser={{database.user}}"
  -            + "&databasePassword={{database.password}}"
  -            + "&databaseDbname=postgres"
  -            + "&databaseServerName=qa"
  -            + "&schemaWhitelist={{database.schema}}"
  -            + "&tableWhitelist={{database.schema}}.question,{{database.schema}}.answer"
  -            + "&offsetStorage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore")
  -            .routeId(QaDatabaseUserNotifier.class.getName() + ".DatabaseReader")         // (1)
  -            .log(LoggingLevel.DEBUG, "Incoming message ${body} with headers ${headers}")
  -            .choice()                                                                    // (2)
  -                .when(isQuestionEvent)
  -                    .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)
  -                    .endChoice()
  -                .when(isAnswerEvent)
  -                    .filter(isCreateOrUpdateEvent)
  -                        .convertBodyTo(Answer.class)
  -                        .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();
  -
  -
  -
  -    @Converter
  -    public static Question questionFromStruct(Struct struct) {                           // (4)
  -        return new Question(struct.getInt64("id"), struct.getString("text"), struct.getString("email"));
  -    }
  -
  -    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);
  -
  +        + "databaseHostname={{database.hostname}}"
  +        + "&databasePort={{database.port}}"
  +        + "&databaseUser={{database.user}}"
  +        + "&databasePassword={{database.password}}"
  +        + "&databaseDbname=postgres"
  +        + "&databaseServerName=qa"
  +        + "&schemaWhitelist={{database.schema}}"
  +        + "&tableWhitelist={{database.schema}}.question,{{database.schema}}.answer"
  +        + "&offsetStorage=org.apache.kafka.connect.storage.MemoryOffsetBackingStore")
  +        .routeId(QaDatabaseUserNotifier.class.getName() + ".DatabaseReader")     // (1)
  +        .log(LoggingLevel.DEBUG, "Incoming message ${body} with headers ${headers}")
  +        .choice()                                                                // (2)
  +          .when(isQuestionEvent)
  +            .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)
  +            .endChoice()
  +          .when(isAnswerEvent)
  +            .filter(isCreateOrUpdateEvent)
  +              .convertBodyTo(Answer.class)
  +              .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.
  
positive.code: 
  To run the demo successfully you need to have a developer Twitter account with appropriate API keys and secrets.
  
  Got to the application directory and build all components
  
  [source,shell]
  ----
  $ mvn clean install
  ----
  
  Start PostgreSQL database and MailHog SMTP server
  
  [source,shell]
  ----
  $ mvn -N docker:start
  ----
  
  In another terminal start the Q&A application backend service
  
  [source,shell]
  ----
  $ java -jar qa-app/target/component-qa-app-1.0.0-SNAPSHOT-runner.jar
  ----
  
  In yet another terminal start the Camel pipeline (provide your own Twitter API credentials)
  
  [source,shell]
  ----
  $ mvn exec:java -Dtwitter.consumerKey=<...> -Dtwitter.consumerSecret=<...> -Dtwitter.accessToken=<...> -Dtwitter.accessTokenSecret=<...> -pl qa-camel
  ----
  
  Create a question and three answers to it
  [source,shell]
  ----
  $ curl -v -X POST -H 'Content-Type: application/json' http://0.0.0.0:8080/question/ -d @src/test/resources/messages/create-question.json
  
negative.code: 
  [source,xml]
  ----
  <dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>org.apache.camel</groupId>
        <artifactId>camel-bom</artifactId>
        <version>${version.camel}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
  
      <!-- Use required Debezium version -->
      <dependency>
        <groupId>io.debezium</groupId>
        <artifactId>debezium-connector-postgres</artifactId>
        <version>${version.debezium}</version>
      </dependency>
      <dependency>
        <groupId>io.debezium</groupId>
        <artifactId>debezium-embedded</artifactId>
        <version>${version.debezium}</version>
      </dependency>
      <dependency>
        <groupId>io.debezium</groupId>
        <artifactId>debezium-core</artifactId>
        <version>${version.debezium}</version>
      </dependency>
  
    </dependencies>
  </dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-core</artifactId>
    </dependency>
    <dependency>
      <groupId>org.apache.camel</groupId>
      <artifactId>camel-debezium-postgres</artifactId>
    </dependency>
  </dependencies>
  ----
  
  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.
  