anchor.diff:  
   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
  +Go to the application directory and build all components:
   
   [source,shell]
   ----
   $ mvn clean install
   ----
   
  -Start PostgreSQL database and MailHog SMTP server
  +Start PostgreSQL database and MailHog SMTP server:
   
   [source,shell]
   ----
   $ mvn -N docker:start
   ----
   
  -In another terminal start the Q&A application backend service
  +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)
  +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
  +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
  
positive.code: 
  [source,java]
  ----
      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);
  
  ----
  
  1. `from` is the Debezium source endpoint. The URI parts map directly to connector configuration options.
  
negative.code: There are multiple ways how to achieve this goal
  
  * Write data to Kafka using Debezium and follow with a combination of Kafka Streams pipelines and Kafka Connect connectors to deliver the changes to other systems
  * Use https://debezium.io/documentation/reference/1.1/operations/embedded.html[Debezium Embedded engine] in a Java standalone application and write the integration code using plain Java; that's often used to send change events to alternative messaging infrastructure such as Amazon Kinesis, Google Pub/Sub etc.
  * Use an existing integration framework or service bus to express the pipeline logic
  
  This article is focusing on the third option - a dedicated integration framework.
  
  
  == Apache Camel
  
  https://camel.apache.org/[Camel] is an open-source integration framework that enables the developer to read, transform, route and write data from and to disparate systems and services.
  It provides a large amount of https://camel.apache.org/components/latest/[components] that either provide interfaces to 3rd party systems or offers an implementation of https://en.wikipedia.org/wiki/Enterprise_Integration_Patterns[enterprise integration patterns].
  
  This combination allows the developer to easily connect to target systems and express integration pipelines using a declarative DSL.
  
  
  === Camel and Debezium
  
  https://camel.apache.org/blog/Camel3-Whatsnew/[Camel 3] has been released by the end of the year 2019 and in addition to the major re-architecture new Debezium components have been added to the codebase.
  It also enables Camel to be used as https://camel.apache.org/camel-kafka-connector/latest/index.html[a connector] in Kafka Connect runtime.
  
  This post focuses solely on Debezium components use and the latter option will be covered in a future post. 
  
  * https://camel.apache.org/components/latest/debezium-mysql-component.html[Debezium MySQL Connector Component]
  * https://camel.apache.org/components/latest/debezium-postgres-component.html[Debezium PostgreSQL Connector Component]
  * https://camel.apache.org/components/latest/debezium-sqlserver-component.html[Debezium SQL Server Connector Component]
  * https://camel.apache.org/components/latest/debezium-mongodb-component.html[Debezium MongoDB Connector Component]
  
  As you can see every _non-incubating_ Debezium connector is represented by its dedicated component.
  The advantage of this solution is the complete isolation of dependencies and the type-safe configuration of connector instances.
  
  Internally the component exposes a Debezium https://camel.apache.org/manual/latest/endpoint.html[endpoint] with Camel event driven https://www.javadoc.io/doc/org.apache.camel/camel-api/latest/org/apache/camel/Consumer.html[consumer] that encapsulates an instance of https://debezium.io/documentation/reference/1.1/operations/embedded.html[Debezium embedded engine].
  
  
  == Example
  
  We have prepared https://github.com/debezium/debezium-examples/tree/master/camel-component[an example] of how to deploy an integration pipeline using the Camel Debezium component.
  
  The source of change data is a simple Question & Answer (Q & A) application.
  The changes generated by the application (e.g. a new question or answer created) are passed to the Camel pipeline that sends emails via an SMTP server and posts a tweet on a provided Twitter account.
  
  
  === Topology
  