anchor.diff:  
   The third snapshot mode instructs the connector to *never* performs snapshots. When a new connector is configured this way, if will either continue streaming changes from a previous stored offset or it will start from the point in time when the PostgreSQL logical replication slot was first created on the server. Note that this mode is useful only when you know all data of interest is still reflected in the WAL.
   
  -The final snapshot mode, *initial only*, will perform a database snapshot and then stop before streaming any other changes. If the connector had started but did not complete a snapshot before stopping, the connector will restart the snapshot process and stop once the snapshot completes.
  -
  +The fourth snapshot mode, *initial only*, will perform a database snapshot and then stop before streaming any other changes. If the connector had started but did not complete a snapshot before stopping, the connector will restart the snapshot process and stop once the snapshot completes.
  +
  +The final snapshot mode, *custom*, allows the user to inject their own implementation of the `io.debezium.connector.postgresql.spi.Snapshotter` interface via the `snapshot.custom.class` configuration property, with the class on the classpath of your Kafka Connect cluster (or included in the JAR if using the `EmbeddedEngine`). For more details, see the link:#custom-snapshot[Custom Snapshot] section.
  +
  +[[custom-snapshot]]
  +=== Custom Snapshotter SPI
  +For more advanced usages, the user can provide an implementation of the `io.debezium.connector.postgresql.spi.Snapshotter` interface. This interfaces allows control of most of the aspects of how snapshots operate, such as whether to take a snapshot or not and the way the options used to open the snapshot transaction or take locks.
  +
  +The full API of the interface can be seen here:
  +[source,Java,indent=0]
  +----
  +**
  + * This interface is used to determine details about the snapshot process:
  + *
  + * Namely:
  + * - Should a snapshot occur at all
  + * - Should streaming occur
  + * - What queries should be used to snapshot
  + *
  + * While many default snapshot modes are provided with debezium (see documentation for details)
  + * a custom implementation of this interface can be provided by the implementor which
  + * can provide more advanced functionality, such as partial snapshots
  + *
  + * Implementor's must return true for either {@link #shouldSnapshot()} or {@link #shouldStream()}
  + * or true for both.
  + */
  +@Incubating
  +public interface Snapshotter {
  +
  +    void init(PostgresConnectorConfig config, OffsetState sourceInfo,
  +              SlotState slotState);
  +
  +    /**
  +     * @return true if the snapshotter should take a snapshot
  +     */
  +    boolean shouldSnapshot();
  +    /**
  +     * @return true if the snapshotter should stream after taking a snapshot
  +     */
  +    boolean shouldStream();
  +
  +    /**
  +     * @return true if when creating a slot, a snapshot should be exported, which
  +     * can be used as an alternative to taking a lock
  +     */
  +    default boolean exportSnapshot() {
  +        return false;
  +    }
  +
  +    /**
  +     * Generate a valid postgres query string for the specified table, or an empty {@link Optional}
  +     * to skip snapshotting this table (but that table will still be streamed from)
  +     *
  +     * @param tableId the table to generate a query for
  +     * @return a valid query string, or none to skip snapshotting this table
  +     */
  +    Optional<String> buildSnapshotQuery(TableId tableId);
  +
  +    /**
  +     * Return a new string that set up the transaction for snapshotting
  +     *
  +     * @param newSlotInfo if a new slow was created for snapshotting, this contains information from
  +     *                    the `create_replication_slot` command
  +     */
  +    default String snapshotTransactionIsolationLevelStatement(SlotCreationResult newSlotInfo) {
  +        // we're using the same isolation level that pg_backup uses
  +        return "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE;";
  +    }
  +
  +    /**
  +     * Returns a SQL statement for locking the given tables during snapshotting, if required by the specific snapshotter
  +     * implementation.
  +     */
  +    default Optional<String> snapshotTableLockingStatement(Duration lockTimeout, Set<TableId> tableIds) {
  +        String lineSeparator = System.lineSeparator();
  +        StringBuilder statements = new StringBuilder();
  +        statements.append("SET lock_timeout = ").append(lockTimeout.toMillis()).append(";").append(lineSeparator);
  +        // we're locking in SHARE UPDATE EXCLUSIVE MODE to avoid concurrent schema changes while we're taking the snapshot
  +        // this does not prevent writes to the table, but prevents changes to the table's schema....
  +        // DBZ-298 Quoting name in case it has been quoted originally; it doesn't do harm if it hasn't been quoted
  +        tableIds.forEach(tableId -> statements.append("LOCK TABLE ")
  +                .append(tableId.toDoubleQuotedString())
  +                .append(" IN SHARE UPDATE EXCLUSIVE MODE;")
  +                .append(lineSeparator));
  +        return Optional.of(statements.toString());
  +    }
  +}
  +----
  +
  +All of the builtin snapshot modes are implemented in terms of this interface as well.
   
   [[streaming-changes]]
   === Streaming Changes
  
positive.code: [[snapshots]]
  === Snapshots
  
  Most PostgreSQL servers are configured to not retain the complete history of the database in the WAL segments, so the PostgreSQL connector would be unable to see the entire history of the database by simply reading the WAL. So, by default the connector will upon first startup perform an initial _consistent snapshot_ of the database. Each snapshot consists of the following steps:
  
  1. Start a transaction with a https://www.postgresql.org/docs/9.6/static/sql-set-transaction.html[SERIALIZABLE, READ ONLY, DEFERRABLE] isolation level to ensure that all subsequent reads within this transaction are done against a single consistent version of the data. Any changes to the data due to subsequent `INSERT`, `UPDATE`, and `DELETE` operations by other clients will not be visible to this transaction.
  2. Obtain a `SHARE UPDATE EXCLUSIVE MODE` lock on each of the monitored tables to ensure that no structural changes can occur to any of the tables while the snapshot is taking place. Note that these locks do not prevent table `INSERTS`, `UPDATES` and `DELETES` from taking place during the operation.
  
negative.code: 
  |`snapshot.mode`
  |`initial`
  |Specifies the criteria for running a snapshot upon startup of the connector. The default is *initial*, and specifies the connector can run a snapshot only when no offsets have been recorded for the logical server name. The *always* option specifies that the connector run a snapshot each time on startup. The *never* option specifies that the connect should never use snapshots and that upon first startup with a logical server name the connector should read from either from where it last left off (last LSN position) or start from the beginning from the point of the view of the logical replication slot. The *initial_only* option specifies that the connector should only take an initial snapshot and then stop, without processing any subsequent changes. Finally, if set to *custom* then the user must also set `snapshot.custom.class` which is a custom implementation of the `io.debezium.connector.postgresql.spi.Snapshotter` interface. See link:#snapshots[snaphosts]
  
  |`snapshot.custom.class`
  |
  | A full java class name that must be an implementation of the `io.debezium.connector.postgresql.spi.Snapshotter` interface. Only used when `snapshot.mode` is *custom*
  
  |`snapshot.lock.timeout.ms`
  |`10000`
  