anchor.diff:                                            .using(config)
                                             .notifying(this::handleEvent)
                                             .build();
  -    
  +
       // Run the engine asynchronously ...
       Executor executor = ...
       executor.execute(engine);
  
positive.code:                                         .with("database.password", "mysqlpw")
                                          .with("server.id", 85744)
                                          .with("database.server.name", "products")
                                          .with("database.history",  
                                                "io.debezium.relational.history.FileDatabaseHistory")
                                          .with("database.history.file.filename",  
                                                "/path/to/storage/dbhistory.dat")
                                          .build())
  ----
  
negative.code:    // Define the configuration for the embedded and MySQL connector ...
      Configuration config = Configuration.create()
                                          /* begin engine properties */
                                          .with("connector.class",
                                                "io.debezium.connector.mysql.MySqlConnector")
                                          .with("offset.storage",
                                                "org.apache.kafka.connect.storage.FileOffsetBackingStore")
                                          .with("offset.storage.file.filename",
                                                "/path/to/storage/offset.dat")
                                          .with("offset.flush.interval.ms", 60000);
  ----
  
  This creates a new `Configuration` object and uses a fluent-style builder API to set several fields required by the engine regardless of which connector is being used. The first is a name for the engine that will be used within the source records produced by the connector and its internal state, so use something meaningful in your application. The `connector.class` field defines the name of the class that extends the Kafka Connect `org.apache.kafka.connect.source.SourceConnector` abstract class; in this example, we specify Debezium's `MySqlConnector` class.
  