anchor.diff:  
   * xref:configuring-a-replication-slot-for-the-debezium-pgoutput-plug-in[]
   * xref:setting-up-postgresql-permissions-required-by-debezium-connectors[]
  -* xref:[]
  +* xref:configuring-postgresql-to-manage-debezium-wal-disk-space-consumption[]
  +
  +endif::product[]
  +
  +ifdef::product[]
   
   // Type: concept
   // ModuleID: configuring-a-replication-slot-for-the-debezium-pgoutput-plug-in
  
positive.code: 
  |[[postgresql-property-snapshot-mode]]<<postgresql-property-snapshot-mode, `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. The *exported* option specifies that the database snapshot will be based on the point in time when the replication slot was created and is an excellent way to perform the snapshot in a lock-free way.
  ifdef::community[]
  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-prefix}:{link-postgresql-connector}#postgresql-snapshots[snapshots].
  
  |[[postgresql-property-snapshot-custom-class]]<<postgresql-property-snapshot-custom-class, `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
  endif::community[]
  
  |[[postgresql-property-snapshot-lock-timeout-ms]]<<postgresql-property-snapshot-lock-timeout-ms, `snapshot.lock.timeout.ms`>>
  |`10000`
  |Positive integer value that specifies the maximum amount of time (in milliseconds) to wait to obtain table locks when performing a snapshot. If table locks cannot be acquired in this time interval, the snapshot will fail. See {link-prefix}:{link-postgresql-connector}#postgresql-snapshots[snapshots]
  
  |[[postgresql-property-snapshot-select-statement-overrides]]<<postgresql-property-snapshot-select-statement-overrides, `snapshot.select.statement.overrides`>>
  |
  |Controls which rows from tables will be included in snapshot. +
  This property contains a comma-separated list of fully-qualified tables _(DB_NAME.TABLE_NAME)_. Select statements for the individual tables are specified in further configuration properties, one for each table, identified by the id `snapshot.select.statement.overrides.[DB_NAME].[TABLE_NAME]`. The value of those properties is the SELECT statement to use when retrieving data from the specific table during snapshotting. _A possible use case for large append-only tables is setting a specific point where to start (resume) snapshotting, in case a previous snapshotting was interrupted._ +
  NOTE: This setting has impact on snapshots only. Events generated by logical decoder are not affected by it at all.
  
  |[[postgresql-property-event-processing-failure-handling-mode]]<<postgresql-property-event-processing-failure-handling-mode, `event.processing{zwsp}.failure.handling.mode`>>
  |`fail`
  | Specifies how the connector should react to exceptions during processing of events.
  `fail` will propagate the exception (indicating the offset of the problematic event), causing the connector to stop. +
  `warn` will cause the problematic event to be skipped and the offset of the problematic event to be logged. +
  `skip` will cause the problematic event to be skipped.
  
  |[[postgresql-property-max-queue-size]]<<postgresql-property-max-queue-size, `max.queue.size`>>
  |`20240`
  |Positive integer value that specifies the maximum size of the blocking queue into which change events received via streaming replication are placed before they are written to Kafka. This queue can provide backpressure when, for example, writes to Kafka are slower or if Kafka is not available.
  
  |[[postgresql-property-max-batch-size]]<<postgresql-property-max-batch-size, `max.batch.size`>>
  |`10240`
  |Positive integer value that specifies the maximum size of each batch of events that should be processed during each iteration of this connector.
  
  |[[postgresql-property-poll-interval-ms]]<<postgresql-property-poll-interval-ms, `poll.interval.ms`>>
  |`1000`
  |Positive integer value that specifies the number of milliseconds the connector should wait during each iteration for new change events to appear. Defaults to 1000 milliseconds, or 1 second.
  
  |[[postgresql-property-include-unknown-datatypes]]<<postgresql-property-include-unknown-datatypes, `include.unknown.datatypes`>>
  |`false`
  |When {prodname} meets a field whose data type is unknown, then by default the field is omitted from the change event and a warning is logged.
  In some cases it may be preferable though to include the field and send it downstream to clients in the opaque binary representation so the clients will decode it themselves.
  Set to `false` to filter unknown data out of events and `true` to keep them in binary format.
  The exact representation can be controlled via the {link-prefix}:{link-postgresql-connector}#connector-property-binary-handling-mode[binary handling mode] setting
  
  NOTE: The clients risk backward compatibility issues. Not only may the database specific binary representation change between releases, but also when the datatype is supported by {prodname} eventually, it will be sent downstream in a logical type, requiring adjustments by consumers. In general, when encountering unsupported data types, please file a feature request so that support can be added.
  
  |[[postgresql-property-database-initial-statements]]<<postgresql-property-database-initial-statements, `database.initial.statements`>>
  |
  |A semicolon separated list of SQL statements to be executed when a JDBC connection (not the transaction log reading connection) to the database is established.
  Use doubled semicolon (';;') to use a semicolon as a character and not as a delimiter.
  
  NOTE: The connector may establish JDBC connections at its own discretion, so this should typically be used for configuration of session parameters only, but not for executing DML statements.
  
  |[[postgresql-property-heartbeat-interval-ms]]<<postgresql-property-heartbeat-interval-ms, `heartbeat.interval.ms`>>
  |`0`
  |Controls how frequently heartbeat messages are sent. +
  This property contains an interval in milli-seconds that defines how frequently the connector sends messages into a heartbeat topic.
  This can be used to monitor whether the connector is still receiving change events from the database.
  You also should leverage heartbeat messages in cases where only records in non-captured tables are changed for a longer period of time.
  In such situation the connector would proceed to read the log from the database but never emit any change messages into Kafka,
  which in turn means that no offset updates will be committed to Kafka.
  This will cause the WAL files to be retained by the database longer than needed
  (as the connector actually has processed them already but never got a chance to flush the latest retrieved LSN to the database)
  and also may result in more change events to be re-sent after a connector restart.
  Set this parameter to `0` to not send heartbeat messages at all. +
  Disabled by default.
  
  |[[postgresql-property-heartbeat-topics-prefix]]<<postgresql-property-heartbeat-topics-prefix, `heartbeat.topics.prefix`>>
  |`__debezium-heartbeat`
  |Controls the naming of the topic to which heartbeat messages are sent. +
  The topic is named according to the pattern `<heartbeat.topics.prefix>.<server.name>`.
  
  |[[postgresql-property-heartbeat-action-query]]<<postgresql-property-heartbeat-action-query, `heartbeat.action.query`>>
  |
  |If specified, this query will be executed upon every heartbeat against the source database.
  
  This can be used to overcome the situation described in {link-prefix}:{link-postgresql-connector}#postgresql-wal-disk-space[WAL Disk Space Consumption],
  where capturing changes from a low-traffic database on the same host as a high-traffic database prevents {prodname} from processing any WAL records and thus acknowledging WAL positions with the database.
  
  Inserting records into some heartbeat table (which must have been created upfront) will allow the connector to receive changes from the low-traffic database and acknowledge their LSNs,
  preventing an unbounded WAL growth on the database host.
  
  Example: `INSERT INTO test_heartbeat_table (text) VALUES ('test_heartbeat')`
  
  |[[postgresql-property-schema-refresh-mode]]<<postgresql-property-schema-refresh-mode, `schema.refresh.mode`>>
  |`columns_diff`
  |Specify the conditions that trigger a refresh of the in-memory schema for a table.
  
  `columns_diff` (the default) is the safest mode, ensuring the in-memory schema stays in-sync with the database table's schema at all times.
  
  `columns_diff_exclude_unchanged_toast` instructs the connector to refresh the in-memory schema cache if there is a discrepancy between it
  and the schema derived from the incoming message, unless unchanged TOASTable data fully accounts for the discrepancy.
  
  This setting can improve connector performance significantly if there are frequently-updated tables that
  have TOASTed data that are rarely part of these updates. However, it is possible for the in-memory schema to
  become outdated if TOASTable columns are dropped from the table.
  
  |[[postgresql-property-snapshot-delay-ms]]<<postgresql-property-snapshot-delay-ms, `snapshot.delay.ms`>>
  |
  |An interval in milli-seconds that the connector should wait before taking a snapshot after starting up; +
  Can be used to avoid snapshot interruptions when starting multiple connectors in a cluster, which may cause re-balancing of connectors.
  
  |[[postgresql-property-snapshot-fetch-size]]<<postgresql-property-snapshot-fetch-size, `snapshot.fetch.size`>>
  |`10240`
  |Specifies the maximum number of rows that should be read in one go from each table while taking a snapshot.
  The connector will read the table contents in multiple batches of this size. Defaults to 10240.
  
  |[[postgresql-property-slot-stream-params]]<<postgresql-property-slot-stream-params, `slot.stream.params`>>
  |
  |Optional list of parameters to be passed to the configured logical decoding plug-in.
  ifdef::community[]
  Can for instance be used to enable server-side table filtering when using the wal2json plug-in.
  Allowed values depend on the chosen plug-in are separated by semicolon.
  endif::community[]
  For example, `add-tables=public.table,public.table2;include-lsn=true`.
  
  |[[postgresql-property-sanitize-field-names]]<<postgresql-property-sanitize-field-names, `sanitize.field.names`>>
  |`true` when connector configuration explicitly specifies the `key.converter` or `value.converter` parameters to use Avro, otherwise defaults to `false`.
  |Whether field names will be sanitized to adhere to Avro naming requirements.
  See {link-prefix}:{link-avro-serialization}#avro-naming[Avro naming] for more details.
  
  |[[postgresql-property-slot-max-retries]]<<postgresql-property-slot-max-retries, `slot.max.retries`>>
  |6
  |How many times to retry connecting to a replication slot when an attempt fails.
  
  |[[postgresql-property-slot-retry-delay-ms]]<<postgresql-property-slot-retry-delay-ms, `slot.retry.delay.ms`>> +
  |10000 (10 seconds)
  |The number of milli-seconds to wait between retry attempts when the connector fails to connect to a replication slot.
  
  |[[postgresql-property-toasted-value-placeholder]]<<postgresql-property-toasted-value-placeholder, `toasted.value.placeholder`>>
  |`__debezium_unavailable_value`
  |Specify the constant that will be provided by {prodname} to indicate that the original value is a toasted value not provided by the database.
  If starts with `hex:` prefix it is expected that the rest of the string repesents hexadecimally encoded octets.
  See {link-prefix}:{link-postgresql-connector}#postgresql-toasted-values[Toasted Values] for additional details.
  
  |[[postgresql-property-provide-transaction-metadata]]<<postgresql-property-provide-transaction-metadata, `provide.transaction.metadata`>>
  |`false`
  |When set to `true` {prodname} generates events with transaction boundaries and enriches data events envelope with transaction metadata.
  
  See {link-prefix}:{link-postgresql-connector}#postgresql-transaction-metadata[Transaction Metadata] for additional details.
  
  |[[postgresql-property-retriable-restart-connector-wait-ms]]<<postgresql-property-retriable-restart-connector-wait-ms, `retriable.restart{zwsp}.connector.wait.ms`>> +
  |10000 (10 seconds)
  
negative.code: [[postgresql-topic-names]]
  === Topics names
  
  The PostgreSQL connector writes events for all insert, update, and delete operations on a single table to a single Kafka topic. By default, the Kafka topic name is _serverName_._schemaName_._tableName_ where: 
  
  * _serverName_ is the logical name of the connector as specified with the `database.server.name` connector configuration property.
  * _schemaName_ is the name of the database schema where the operation occurred.
  * _tableName_ is the name of the database table in which the operation occurred.
  
  For example, suppose that `fulfillment` is the logical server name in the configuration for a connector that is capturing changes in a PostgreSQL installation that has a `postgres` database and an `inventory` schema that contains four tables: `products`, `products_on_hand`, `customers`, and `orders`. The connector would stream records to these four Kafka topics:
  
  * `fulfillment.inventory.products`
  * `fulfillment.inventory.products_on_hand`
  * `fulfillment.inventory.customers`
  * `fulfillment.inventory.orders`
  
  Now suppose that the tables are not part of a specific schema but were created in the default `public` PostgreSQL schema. The names of the Kafka topics would be:
  
  * `fulfillment.public.products`
  * `fulfillment.public.products_on_hand`
  