<a id="influx-db-source-connector"></a>

# InfluxDB Source Connector [Deprecated] for Confluent Platform

#### IMPORTANT
- This connector is deprecated and reaches end of life (EOL) on the Confluent Platform 8.2 end of support (EOS) date.
  For more information, see [Connector support lifecycle policy](https://docs.confluent.io/platform/current/connect/supported.html#support-policy-for-self-managed-connectors).
- <!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

  Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
  [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
  page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
  are no longer available for download.

The Kafka Connect InfluxDB Source connector allows you to import data from an InfluxDB host into an Apache Kafka® topic.

Data is loaded by periodically executing an Influx query and creating an output
record for each row in the result set. By default, all measurements in a
database are copied, each to its own output topic. The database is monitored for
new measurements and adapts automatically. When copying data from a measurement,
the connector loads only new records.

## Features

The source connector supports copying measurements with a variety of InfluxDB
data types, adding measurements from the database dynamically, whitelists and
blacklists, varying polling intervals, and other settings. However, the most
important features for most users are the settings controlling how data is
incrementally copied from the database.

Kafka Connect tracks the latest record it retrieved from each measurement, so
it can start in the correct location on the next iteration (or in case of a
crash). The source connector uses this functionality to only get updated records
from measurements (or from the output of a custom query) on each iteration.
Several modes are supported, each of which differs in how modified rows are
detected.

The InfluxDB Source connector includes the following features:

- [At least once delivery](#influxdb-source-at-least-once-delivery)
- [Supports one task](#influxdb-source-one-task)
- [Client-side encryption](#influxdb-source-csfle-sm)

<a id="influxdb-source-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records are delivered at least once to the Kafka
topic. If the connector restarts, there may be some duplicate
records in the Kafka topic.

<a id="influxdb-source-one-task"></a>

### Supports one task

The InfluxDB Source connector supports running a single task, which is
initiated when in QUERY mode. Otherwise, the connector will initiate tasks
based on the minimum number of measurements or `max-tasks` configured.

<a id="influxdb-source-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

See [Confluent Platform license](influx_db_source_connector_config.md#influx-db-source-connector-license-config) for license properties and [License topic configuration](influx_db_source_connector_config.md#influx-db-source-license-topic-configuration) for information about the license topic. License details are the same for both the source and sink connectors.

## Configuration Properties

For a complete list of configuration properties for this connector, see [Configuration Reference for InfluxDB Source Connector for Confluent Platform](influx_db_source_connector_config.md#influx-db-source-connector-config).

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## Installe InfluxDB Sink Connector

The InfluxDB Connector is compatible with Confluent Platform version 4.1 (and later).

The InfluxDB connector is compatible with Confluent Platform version 4.1 and later. You can
install this connector by using the [Confluent Hub client installation
instructions](https://docs.confluent.io/home/connect/confluent-hub/client.html) or by
manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- If you want to install the connector using Confluent Marketplace, you must install
  the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This
  is installed by default with Confluent Enterprise.

### Install the connector using the Confluent CLI

To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory
and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-influxdb:latest
```

You can install a specific version by replacing `latest` with a version number
as shown in the following example:

```bash
confluent connect plugin install confluentinc/kafka-connect-influxdb:1.2.7
```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-jms) for your
connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/userguide.html#connect-installing-plugins).

## Quick Start

In this quick start, you copy data from a single measurement from a local
Influx database running on Docker into a Kafka topic.

This example assumes you are running Kafka and Schema Registry locally on the default ports.
It also assumes you have Docker installed and running.

First, bring up the Influx database by running the following Docker command:

```bash
docker run -d -p 8086:8086 --name influxdb-local influxdb:1.7.7
```

This starts the Influx database, and maps it to port 8086 on `localhost`.
By default, the user name and password are blank. The database connection URL is `http://localhost:8086`.

To create sample data in the Influx database, log in to the Docker container using the following command:

```bash
docker exec -it <containerid> bash
```

Once you are in the Docker container, log in to InfluxDB shell:

```bash
influx
```

Your output should resemble:

```bash
Connected to http://localhost:8086 version 1.7.7
InfluxDB shell version: 1.7.7
```

### Create Influx Database and Load Data

1. Create an Influx database with this command:
   ```bash
   > create database testdb;
   ```

   Check for the new database by running:
   ```bash
   > show databases;
   ```
2. In the InfluxDB command prompt, create a measurement and seed it with some data:
   ```bash
   > use testdb;
            Using database testdb
   > INSERT coin,id=1 value=100
   ```

### Start InfluxDB Source Connector

Start the Confluent Platform using the Confluent CLI command below.

```bash
confluent local start
```

Next, create a configuration file for the connector. This file is included with
the connector in  `./etc/kafka-connect-influxdb/influxdb-source-connector.properties`
and contains the following settings:

```bash
name=InfluxDBSourceConnector
connector.class=io.confluent.influxdb.source.InfluxdbSourceConnector
tasks.max=1
topic.prefix=influx_
influxdb.url=http://localhost:8086
influxdb.db=testdb
mode=timestamp
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
```

Load the InfluxDB source connector with this configuration.

```bash
confluent local load influx-source-connector --config etc/kafka-connect-influxdb/influxdb-source-connector.properties
```

To check that it has copied the data that was present when you started Kafka
Connect, start a console consumer, reading from the beginning of the topic:

```bash
kafka-console-consumer --bootstrap-server localhost:9092 --topic influx_coin --from-beginning
```

Your output should resemble:

```bash
{"measurement":"coin","tags":{"id":"1"},"time":"2019-07-24T10:14:04.979737851Z","value":100.0}
```

Note that the default polling interval is five seconds, so it may take a few
seconds to show up. Depending on your expected rate of updates or desired
latency, a smaller poll interval could be used to deliver updates more quickly.

[All the features of Kafka Connect](/kafka-connectors/self-managed/userguide.html), including offset
management and fault tolerance, work with the source connector. You can restart
and kill the processes and they will pick up where they left off, copying only
new data (as defined by the `mode` setting).

## Query Modes

Each incremental query mode tracks a set of fields for each point, which it uses
to keep track of which records have been processed and which records are new.
The `mode` setting controls this behavior and supports the following options:

* **Timestamp**: In this mode, a single record containing a timestamp is used
  to track the last time data was processed and to query only for records that have been added since that time.
* **Custom Query**: The source connector supports using custom queries instead of copying whole
  measurements. With a custom query, one of the other automatic update modes can be used as long
  as the necessary `WHERE` clause can be correctly appended to the query. Alternatively, the
  specified query may handle filtering to new updates itself;
  however, note that no offset tracking will be performed (unlike the automatic modes where
  `timestamp` column values are recorded for each record), so the query
  must track offsets itself.
* **Bulk**: This mode is unfiltered and therefore not incremental at all. It will load all records
  from a measurement on each iteration. This can be useful if you want to periodically dump an entire
  measurement where entries are eventually deleted and the downstream system can safely handle duplicates.

## Configuration

The source connector gives you quite a bit of flexibility regarding where you
can import data from and how that data is imported. The full set of
configuration options are listed in [Configuration Reference for InfluxDB Source Connector for Confluent Platform](influx_db_source_connector_config.md#influx-db-source-connector-config),
but here are a few template configurations that cover some common usage
scenarios.

Use a whitelist to limit changes to a subset of measurements in an Influx
database, using `timestamp`  fields that are standard on all whitelisted
measurements to detect records that have been created. This mode is the most
robust because it can use timestamps to guarantee modifications are not missed
even if the process dies in the middle of a query.

```none
name=influx-whitelist-timestamp-source
connector.class=io.confluent.influxdb.source.InfluxdbSourceConnector
tasks.max=10

influxdb.url=http://localhost:8086
influxdb.db=testdb
influxdb.measurement.whitelist=users,products,transactions
mode=timestamp
topic.prefix=influx_
```

Use a blacklist to exclude measurements from copying data from an Influx
database. It will monitor all measurements except blacklisted measurements.

```none
name=influx-blacklist-timestamp-source
connector.class=io.confluent.influxdb.source.InfluxdbSourceConnector
tasks.max=10

influxdb.url=http://localhost:8086
influxdb.db=testdb
influxdb.measurement.blacklist=users,products
mode=timestamp
topic.prefix=influx_
```

Use a custom query instead of loading measurements, which allows you to load data from multiple measurements.

```none
name=influx-whitelist-timestamp-source
connector.class=io.confluent.influxdb.source.InfluxdbSourceConnector
tasks.max=10

influxdb.url=http://localhost:8086
influxdb.db=testdb
topic.prefix=influx_
query=Your-Custom-query
```