anchor.diff:  * `REPLICATION SLAVE`
   * `REPLICATION CLIENT`
   
  -The first three privileges are required when reading a consistent snapshot of the databases. The last two privileges allowing the database to read the server's binlog that is normally used for MySQL replication.
  +The first three privileges are required when reading a consistent snapshot of the databases. The last two privileges allow the database to read the server's binlog that is normally used for MySQL replication.
   
   The configuration also includes a numeric identifier for the `server.id`. Since MySQL's binlog is part of the MySQL replication mechanism, in order to read the binlog the `MySqlConnector` instance must join the MySQL server group, and that means this server ID must be https://dev.mysql.com/doc/refman/5.7/en/replication-howto-masterbaseconfig.html[unique within all processes that make up the MySQL server group] and is any integer between 1 and 2^32^-1. In our code we set it to a fairly large but somewhat random value we'll use only for our application.
   
   The configuration also specifies a logical name for the MySQL server. The connector includes this logical name within the topic field of every source record it produces, enabling your application to discern the origin of those records. Our example uses a server name of "products", presumably because the database contains product information. Of course, you can name this anything meaningful to your application.
   
  -When the `MySqlConnector` class runs, it reads the MySQL server's binlog, which includes all data changes and schema changes made to the databases hosted by the server. Since all changes to data are structured in terms of the owning table's schema at the time the change was recorded, the connector needs to track all of the schema changes so that it can properly decode the change events. The connector records the schema information so that, should the connector be restarted and resume reading from the last recorded offset, it knows exactly what the database schemas looked like at that offset. How the connector records the database schema history is defined in the last two fields of our configuration, namely that our connector should use the `FileDatabaseHistory` class to store database schema history changes in the `/path/to/storage/dbhistory.dat` file on the local file system (again, this file can be named anything and stored anywhere). 
  +When the `MySqlConnector` class runs, it reads the MySQL server's binlog, which includes all data changes and schema changes made to the databases hosted by the server. Since all changes to data are structured in terms of the owning table's schema at the time the change was recorded, the connector needs to track all of the schema changes so that it can properly decode the change events. The connector records the schema information so that, should the connector be restarted and resume reading from the last recorded offset, it knows exactly what the database schemas looked like at that offset. How the connector records the database schema history is defined in the last two fields of our configuration, namely that our connector should use the `FileDatabaseHistory` class to store database schema history changes in the `/path/to/storage/dbhistory.dat` file on the local file system (again, this file can be named anything and stored anywhere).
   
   Finally the immutable configuration is built using the `build()` method. (Incidentally, rather than build it programmatically, we could have *read* the configuration from a properties file using one of the `Configuration.read(...)` methods.)
   
  
positive.code:                                           .build();
  ----
  
  A fluent-style builder API is used to create an engine that uses our `Configuration` object and that sends all data change records to the `handleEvent(SourceRecord)` method, which can be any method that matches the signature of the `java.util.function.Consumer<SourceRecord>` functional interface, where `SourceRecord` is the `org.apache.kafka.connect.source.SourceRecord` class. Note that your application's handler function should not throw any exceptions; if it does, the engine will log any exception thrown by the method and will continue to operate on the next source record, but your application will not have another chance to handle the particular source record that caused the exception, meaning your application might become inconsistent with the database. 
  
  At this point, we have an existing `EmbeddedEngine` object that is configured and ready to run, but it doesn't do anything. The `EmbeddedEngine` is designed to be executed asynchronously by an `Executor` or `ExecutorService`:
  
  
negative.code: ----
      try {
          while (!connector.await(30, TimeUnit.SECONDS)) {
              logger.info("Wating another 30 seconds for the embedded engine to shut down");
          }
      } catch ( InterruptedException e ) {
          Thread.interrupted();
  