<a id="jdbc-source-configs"></a>

# Configuration Reference for JDBC Source Connector for Confluent Platform

To use this connector, specify the name of the connector class in the
`connector.class` configuration property.

```properties
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
```

Connector-specific configuration properties are described below.

## Database Connection Security

In the connector configuration you will notice there are no security parameters. This is because
SSL is not part of the JDBC standard and will depend on the JDBC driver in use. In general, you
must configure SSL using the `connection.url` parameter. For example, with MySQL it would
look similar to the following:

```properties
connection.url="jdbc:mysql://127.0.0.1:3306/sample?verifyServerCertificate=false&useSSL=true&requireSSL=true"
```

Confluent recommends you view your JDBC driver documentation for support and configuration.

## Database

The JDBC connector allows you to configure any driver parameter using the prefix
`connection.*`. For example:

- For Postgres, you can use the following:
  ```text
  "connection.loginTimeout": "10"
  ```
- For Oracle, you can use the following:
  ```text
  "connection.oracle.jdbc.ReadTimeout": "45000"
  ```

For more details, log in to the [Confluent Support Portal](https://support.confluent.io/) and view [How to set JDBC driver
configuration using JDBC connector](https://support.confluent.io/hc/en-us/articles/15171712953108-How-to-set-JDBC-driver-configuration-using-JDBC-connector?source=search&auth_token=eyJhbGciOiJIUzI1NiJ9.eyJhY2NvdW50X2lkIjo2ODg5NDYsInVzZXJfaWQiOjc2MzA0NTc1OTE1NzIsInRpY2tldF9pZCI6MTU0ODI4LCJjaGFubmVsX2lkIjo2MywidHlwZSI6IlNFQVJDSCIsImV4cCI6MTY4NjIxNDg1OX0.2X6qvtfiDrMvw1nLCcDcDW5lmqMf7Hl-VOYbrjj1Ls0).

`connection.url`
: JDBC connection URL.
  <br/>
  For example: `jdbc:oracle:thin:@localhost:1521:orclpdb1`,
  `jdbc:mysql://localhost/db_name`,
  `jdbc:sqlserver://localhost;instance=SQLEXPRESS;databaseName=db_name`
  <br/>
  * Type: string
  * Importance: high
  * Dependents: `table.whitelist`, `table.blacklist`

`connection.user`
: JDBC connection user.
  <br/>
  * Type: string
  * Default: null
  * Importance: high
  <br/>
    Confluent strongly recommends you follow the principle of least
    privilege when creating the database user for this connector. Permissions
    should be tailored to specific actions on the required tables to ensure the
    connector can only access the data, or perform the actions necessary for its
    function.

`connection.password`
: JDBC connection password.
  <br/>
  * Type: password
  * Default: null
  * Importance: high

`jdbc.credentials.provider.class`
: Credentials provider to use for authentication to configure the database. By default, the connector uses
  `DefaultJdbcCredentialsProvider`. Configure the class with the fully qualified name of your
  custom credentials provider class.
  <br/>
  * Type: class
  * Default: `io.confluent.connect.jdbc.util.DefaultJdbcCredentialsProvider`
  * Valid Values: Any class implementing interface: `io.confluent.connect.jdbc.util.JdbcCredentialsProvider`
  * Importance: low

`connection.attempts`
: Maximum number of attempts to retrieve a valid JDBC connection. Must be a
  positive integer.
  <br/>
  * Type: int
  * Default: 3
  * Valid Values: [1,…]
  * Importance: low

`connection.backoff.ms`
: Backoff time in milliseconds between connection attempts.
  <br/>
  * Type: long
  * Default: 10000
  * Importance: low

`catalog.pattern`
: Catalog pattern to fetch table metadata from the database.
  <br/>
  * Type: string
  * Default: null
    * `""` retrieves those without a catalog
    * null (default) indicates that the schema name is not used to narrow the
      search and that all table metadata is fetched, regardless of the catalog.
  * Importance: medium

`table.whitelist`
: List of tables to include in copying. If specified, `table.blacklist` may
  not be set. Use a comma-separated list to specify multiple tables (for
  example, `table.whitelist: "User, Address, Email"`).
  <br/>
  * Type: list
  * Default: “”
  * Importance: medium

`table.blacklist`
: List of tables to exclude from copying. If specified, `table.whitelist` may
  not be set. Use a comma-separated list to specify multiple tables (for
  example, `table.blacklist: "User, Address, Email"`).
  <br/>
  * Type: list
  * Default: “”
  * Importance: medium

`schema.pattern`
: Schema pattern to fetch table metadata from the database.
  <br/>
  * Type: string
  * Default: null
    * `""` retrieves those without a schema.
    * null (default) indicates that the schema name is not used to narrow the
      search and that all table metadata is fetched, regardless of the schema.
  * Importance: high
  <br/>
    #### IMPORTANT
    If you leave this at the default null setting, the connector may time out
    and fail because of the large amount of table metadata being received.
    Make sure to set this parameter for large databases.

`numeric.precision.mapping`
: Whether or not to attempt mapping NUMERIC values by precision to integral
  types. This option is now deprecated. A future version may remove it
  completely. You should use `numeric.mapping` instead.
  <br/>
  * Type: boolean
  * Default: false
  * Importance: low

`numeric.mapping`
: Map NUMERIC values by precision and optionally scale to integral or decimal
  types.
  <br/>
  * Type: string
  * Default: null
  * Valid Values: [none, precision_only, best_fit, best_fit_eager_double]
    * Use `none` if all NUMERIC columns are to be represented by Connect’s
      DECIMAL logical type.
    * Use `best_fit` if NUMERIC columns should be cast to Connect’s INT8,
      INT16, INT32, INT64, or FLOAT64 based upon the column’s precision and scale.This option may
      still represent the NUMERIC value as Connect DECIMAL if it cannot be cast to a native
      type without losing precision. For example, a NUMERIC(20) type with precision 20 would
      not be able to fit in a native INT64 without overflowing and thus would be retained as
      DECIMAL.
    * Use `best_fit_eager_double` if, in addition to the properties of `best_fit` described
      above, it is desirable to always cast NUMERIC columns with scale to Connect FLOAT64 type,
      despite potential of loss in accuracy.
    * Use `precision_only` to map NUMERIC columns based only on the column’s
      precision assuming that column’s scale is 0.
    * The `none` option is the default, but may lead to serialization issues
      with Avro since Connect’s DECIMAL type is mapped to its binary
      representation. `best_fit` is often preferred since it maps to the most
      appropriate primitive type.
  * Importance: low

`dialect.name`
: The name of the database dialect that should be used for this connector. By
  default this is empty, and the connector automatically determines the dialect
  based upon the JDBC connection URL. Use this if you want to override that
  behavior and use a specific dialect. All properly-packaged dialects in the
  JDBC connector plugin can be used.
  <br/>
  * Type: string
  * Default: “”
  * Valid Values: [`, Db2DatabaseDialect, MySqlDatabaseDialect,
    SybaseDatabaseDialect, GenericDatabaseDialect, OracleDatabaseDialect`
    `,SqlServerDatabaseDialect, PostgreSqlDatabaseDialect,
    SqliteDatabaseDialect, DerbyDatabaseDialect` `, SapHanaDatabaseDialect,
    MockDatabaseDialect, VerticaDatabaseDialect`]
  * Importance: low

## Mode

<a id="jdbc-source-configs-mode"></a>

`mode`
: The mode for updating a table each time it is polled. Options include:
  <br/>
  * Type: string
  * Default: “”
  * Valid Values: [, bulk, timestamp, incrementing, timestamp+incrementing]
    * bulk: perform a bulk load of the entire table each time it is polled
    * incrementing: use a strictly incrementing column on each table to detect
      only new rows. Note that this will not detect modifications or deletions of
      existing rows.
    * timestamp: use a timestamp (or timestamp-like) column to detect new and
      modified rows. This assumes the column is updated with each write, and that
      values are monotonically incrementing, but not necessarily unique.
    * timestamp+incrementing: use two columns, a timestamp column that detects
      new and modified rows and a strictly incrementing column which provides a
      globally unique ID for updates so each row can be assigned a unique stream
      offset.
  * Importance: high
  * Dependents: `incrementing.column.name`, `timestamp.column.name`, `validate.non.null`

`incrementing.column.name`
: The name of the strictly incrementing column to use to detect new rows. Any
  empty value indicates the column should be autodetected by looking for an
  auto-incrementing column. This column may not be nullable.
  <br/>
  * Type: string
  * Default: “”
  * Importance: medium

`timestamp.column.name`
: Comma-separated list of one or more timestamp columns to detect new or
  modified rows using the COALESCE SQL function. Rows whose first non-null
  timestamp value is greater than the largest previous timestamp value seen will
  be discovered with each poll. At least one column should not be nullable.
  <br/>
  * Type: string
  * Default: “”
  * Importance: medium

`timestamp.initial`
: The epoch timestamp (in milliseconds) used for initial queries that use
  timestamp criteria. Use -1 to use the current time. If not specified, all data
  will be retrieved.
  <br/>
  * Type: long
  * Default: null
  * Importance: low

`validate.non.null`
: By default, the JDBC connector will validate that all incrementing and
  timestamp tables have NOT NULL set for the columns being used as their
  ID/timestamp. If the tables don’t, JDBC connector will fail to start. Setting
  this to false will disable these checks.
  <br/>
  * Type: boolean
  * Default: true
  * Importance: low

`query`
: If specified, `query` can select new or updated rows. Use this
  setting if you want to join tables, select subsets of columns in a table, or
  filter data. If used, the connector copies data using this query and
  whole-table copying is disabled.
  <br/>
  * Type: string
  * Default: “”
  * Importance: medium

`quote.sql.identifiers`
: When to quote table names, column names, and other identifiers in SQL
  statements. For backward compatibility, the default is `always`.
  <br/>
  * Type: string
  * Default: always
  * Importance: medium

`query.suffix`
: Suffix to append at the end of the generated query.
  <br/>
  * Type: string
  * Default: “”
  * Importance: low

`query.retry.attempts`
: The number of times to retry SQL exceptions encountered when executing queries.
  <br/>
  * Type: int
  * Default: -1
  * Importance: low

`transaction.isolation.mode`
: The mode to control which transaction isolation level is used when running queries
  against the database. By default, no explicit transaction isolation mode is
  set. `SQL_SERVER_SNAPSHOT` will only work against a connector configured to
  write to SQL Server. Options include: `DEFAULT`, `READ_UNCOMMITTED`,
  `READ_COMMITTED`, `REPEATABLE_READ`, `SERIALIZABLE`, and
  `SQL_SERVER_SNAPSHOT`.
  <br/>
  * Type: string
  * Default: DEFAULT
  * Valid Values: [`DEFAULT`, `READ_UNCOMMITED`, `READ_COMMITED`, `REPEATABLE_READ`, `SERIALIZABLE`, `SQL_SERVER_SNAPSHOT`]
  * Importance: low

## Connector

`table.types`
: By default, the JDBC connector will only detect tables with type TABLE from
  the source Database. This config allows a comma-separated list of table
  types to extract.
  <br/>
  * Type: list.
    * TABLE
    * VIEW
    * SYSTEM TABLE
    * GLOBAL TEMPORARY
    * LOCAL TEMPORARY
    * ALIAS
    * SYNONYM
  <br/>
    In most cases it only makes sense to have either TABLE or VIEW.
  * Default: TABLE
  * Importance: low

`poll.interval.ms`
: Frequency in ms to poll for new data in each table.
  <br/>
  * Type: int
  * Default: 5000
  * Importance: high

`poll.linger.ms`
: The maximum time to wait for a record before returning an empty
  batch.
  <br/>
  * Type: long
  * Default: 5000
  * Importance: low

`batch.max.rows`
: Maximum number of rows to include in a single batch when polling for new data.
  This setting can be used to limit the amount of data buffered internally in
  the connector.
  <br/>
  * Type: int
  * Default: 100
  * Importance: low

`table.poll.interval.ms`
: Frequency in ms to poll for new or removed tables, which may result in updated
  task configurations to start polling for data in added tables or stop polling
  for data in removed tables.
  <br/>
  * Type: long
  * Default: 60000
  * Importance: low

`topic.prefix`
: Prefix to prepend to table names to generate the name of the Apache Kafka® topic to
  publish data to, or in the case of a custom query, the full name of the topic
  to publish to.
  <br/>
  * Type: string
  * Importance: high

`timestamp.delay.interval.ms`
: After a row with a certain timestamp appears, this is the amount of time the connector waits to
  include it in the result. You can add a delay to allow transactions with an
  earlier timestamp to complete. The first execution fetches all available
  records (starting at timestamp 0) until the current time minus the delay. Each
  subsequent execution retrieves data from the last time you fetched until the
  current time minus the delay.
  <br/>
  * Type: long
  * Default: 0
  * Importance: high

`db.timezone`
: The name of the JDBC timezone used in the connector when querying with
  time-based criteria. This should be the timezone of the database as well as
  all the columns being queried. The value set in `db.timezone` will also be
  used for timestamp columns while pushing the records to Kafka. Defaults to UTC.
  <br/>
  * Type: string
  * Default: UTC
  * Valid Values: Any valid JDK time zone
  * Importance: medium

`timestamp.granularity`
: Define the granularity of the Timestamp column. Options include:
  <br/>
  * Type: string
  * Default: connect_logical
  * Valid Values: [connect_logical, nanos_long, nanos_string, nanos_iso_datetime_string]
    * connect_logical (default): represents timestamp values using Kafka Connect’s built-in representations
    * micros_long: represents timestamp values as micros since epoch
    * micros_string: represents timestamp values as micros since epoch in string
    * micros_iso_datetime_string: uses the iso format ‘yyyy-MM-dd’T’HH:mm:ss.SSSSSS’
    * nanos_long: represents timestamp values as nanos since epoch
    * nanos_string: represents timestamp values as nanos since epoch in string
    * nanos_iso_datetime_string: uses the iso format ‘yyyy-MM-dd’T’HH:mm:ss.SSSSSSSSS’
  * Importance: low

## Auto topic creation

For more information about Auto topic creation, see [Configuring
Auto Topic Creation for Source
Connectors](/kafka-connectors/self-managed/userguide.html#connect-source-auto-topic-creation).

Configuration properties accept regular expressions (regex) that are defined as
[Java regex](https://docs.oracle.com/javase/8/docs/api/java/util/regex/Pattern.html).

`topic.creation.groups`
: A list of group aliases that are used to define per-group topic configurations for matching topics. A `default` group always exists and matches all topics.
  <br/>
  * Type: List of String types
  * Default: empty
  * Possible Values: The values of this property refer to any additional groups. A `default` group is always defined for topic configurations.

`topic.creation.$alias.replication.factor`
: The replication factor for new topics created by the connector. This value must not be larger than the number of brokers in the Kafka cluster. If this value is larger than the number of Kafka brokers, an error occurs when the connector attempts to create a topic. This is a **required property** for the `default` group. This property is optional for any other group defined in `topic.creation.groups`. Other groups use the Kafka broker default value.
  <br/>
  * Type: int
  * Default: n/a
  * Possible Values: `>= 1` for a specific valid value or `-1` to use the Kafka broker’s default value.

`topic.creation.$alias.partitions`
: The number of topic partitions created by this connector. This is a **required property** for the `default` group. This property is optional for any other group defined in `topic.creation.groups`. Other groups use the Kafka broker default value.
  <br/>
  * Type: int
  * Default: n/a
  * Possible Values: `>= 1` for a specific valid value or `-1` to use the Kafka broker’s default value.

`topic.creation.$alias.include`
: A list of strings that represent regular expressions that match topic names. This list is used to include topics with matching values, and apply this group’s specific configuration to the matching topics. `$alias` applies to any group defined in `topic.creation.groups`. This property does not apply to the `default` group.
  <br/>
  * Type: List of String types
  * Default: empty
  * Possible Values: Comma-separated list of exact topic names or regular expressions.

`topic.creation.$alias.exclude`
: A list of strings representing regular expressions that match topic names. This list is used to exclude topics with matching values from getting the group’s specfic configuration. `$alias` applies to any group defined in `topic.creation.groups`. This property does not apply to the `default` group. Note that exclusion rules override any inclusion rules for topics.
  <br/>
  * Type: List of String types
  * Default: empty
  * Possible Values: Comma-separated list of exact topic names or regular expressions.

`topic.creation.$alias.${kafkaTopicSpecificConfigName}`
: Any of the [Changing Broker Configurations Dynamically](/platform/current/kafka/dynamic-config.html#changing-broker-configurations-dynamically) for the version of the Kafka broker where the records will be written. The broker’s topic-level configuration value is used if the configuration is not specified for the rule. `$alias` applies to the `default` group as well as any group defined in `topic.creation.groups`.
  <br/>
  * Type: property values
  * Default: Kafka broker value

<!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

## CSFLE and CSPE configurations

`csfle.enabled`

Accepts a boolean value. CSFLE is enabled for the connector if `csfle.enabled` is set to True.

* Type: boolean
* Default: False

`auto.register.schemas`

Specifies if the Serializer should attempt to register the Schema with Schema Registry.

* Type: boolean
* Default: true
* Importance: medium

`use.latest.version`

Only applies when `auto.register.schemas` is set to false. If `auto.register.schemas` is set to false and `use.latest.version` is set to true, then instead of deriving a schema for the object passed to the client for serialization, Schema Registry uses the latest version of the schema in the subject for serialization.

* Type: boolean
* Default: true
* Importance: medium