Skip to main content

Kafka

Availability
Available CoreNot available StandardNot available PlusNot available ProNot available Enterprise FlexAvailable Self-Managed EnterpriseAvailable PyAirbyte
Support Level
Marketplace
Connector Version
0.4.2 (last updated 7 months ago)
Sync Success Rate
Usage Rate
Data workers used
0.5 per sync (Learn more)
Definition ID
d917a47b-8537-4d0d-8c10-36a9928d4265

This page guides you through the process of setting up the Kafka source connector.

Set up guide​

Step 1: Set up Kafka​

To use the Kafka source connector, you'll need:

  • A Kafka cluster 1.0 or above
  • Airbyte user should be allowed to read messages from topics, and these topics should be created before reading from Kafka.

Step 2: Setup the Kafka source in Airbyte​

You'll need the following information to configure the Kafka source:

  • Bootstrap Servers - A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.
  • MessageFormat - How to deserialize messages; choose either:
    • JSON
    • AVRO if choosing avro, several other options are available
      • Deserialization strategy - To deserialize Avro binary, set the [subject name strategy][subject-name-strategy] for choosing the schema from the registry
      • Schema Registry URL - Host/port to connect to the schema registry server
      • Username and password, optional to authenticate to the schema registry server
  • Protocol - The Protocol used to communicate with brokers. If using SASL, this includes how to authenticate to the brokers.
    • PLAINTEXT - no authentication and no encryption
    • SASL PLAINTEXT - no encryption; a SASL JAAS Config is needed to authenticate
    • SASL SSL - encrypted
      • SASL JAAS Config configures authentication to the brokers
      • SASL Mechanism appropriate to your brokers' auth mechanism
      • OAUTHBEARER token endpoint URL Optional; If using SASL Mechanism OAUTHBEARER, set to the token endpoint URL; note this does not apply to schema registry auth
  • Subscription Method - You can choose to manually assign a list of partitions, or subscribe to all topics matching specified pattern to get dynamically assigned partitions.
    • List of topic - comma separated topic:partitions, each partition becomes a stream in the connection
    • Topic pattern - a pattern to use, each matching topic becomes a stream in the connection

Some optional fields that are recommended to set according to your use

  • Client ID - An ID string to pass to the brokers when making requests. The purpose of this is to be able to track the source of requests beyond just ip/port by allowing a logical application name to be included in server-side request logging. (e.g. airbyte-consumer)
  • Group ID - The Group ID is how you distinguish different consumer groups. (e.g. group.id)
  • Test Topic - The Topic to test whether Airbyte can consume messages. (e.g. test.topic)
  • Polling Time - Amount of time in milliseconds Kafka connector should try to poll for messages, per sync

For Airbyte Open Source​

  1. Go to the Airbyte UI and in the left navigation bar, click Sources. In the top-right corner, click +new source.
  2. On the Set up the source page, enter the name for the Kafka connector and select Kafka from the Source type dropdown.
  3. Follow the Setup the Kafka source in Airbyte

Supported sync modes​

The Kafka source connector supports the following sync modes:

FeatureSupported?(Yes/No)Notes
Full Refresh SyncYes
Incremental - Append SyncYes
NamespacesNo

Supported Format​

JSON - Json value messages. It does not support schema registry now.

AVRO - deserialize Using confluent API. Please refer (https://docs.confluent.io/platform/current/schema-registry/serdes-develop/serdes-avro.html)

IP allow list​

If you use Airbyte Cloud and your organization restricts access to specific IPs, add the Airbyte Cloud IP addresses to your allow list.

Reference​

Config fields reference

Field
Type
Property name
string
bootstrap_servers
object
protocol
object
subscription
object
MessageFormat
integer
auto_commit_interval_ms
string
auto_offset_reset
string
client_dns_lookup
string
client_id
boolean
enable_auto_commit
string
group_id
integer
max_poll_records
integer
max_records_process
integer
polling_time
integer
receive_buffer_bytes
integer
repeated_calls
integer
request_timeout_ms
integer
retry_backoff_ms
string
test_topic

Changelog​

Expand to review
VersionDatePull RequestSubject
0.4.22025-07-1062919Convert to new gradle build flow
0.4.12025-04-3059171Fixing vulnerabilities in dependencies.
0.4.02025-03-1855828Add configurations for AWS MSK IAM and fix JSON format
0.3.02025-02-1853231Add configurations for OAUTHBEARER SASL Mechanism
0.2.82025-02-0753221For AVRO MessageFormat, schema_registry_password is a secret
0.2.72025-01-1051480Use a non root base image
0.2.62024-12-1849907Use a base image: airbyte/java-connector-base:1.0.0
0.2.52024-06-1232538Fix empty airbyte data column
0.2.42024-02-1335229Adopt CDK 0.20.4
0.2.42024-01-2434453bump CDK version
0.2.32022-12-0619587Fix missing data before consumer is closed
0.2.22022-11-0418648Add missing record_count increment for JSON
0.2.12022-11-04This version was the same as 0.2.0 and was committed so using 0.2.2 next to keep versions in order
0.2.02022-08-2213864Added AVRO format support and Support for maximum records to process
0.1.72022-06-1713864Updated stacktrace format for any trace message errors
0.1.62022-05-2912903Add Polling Time to Specification (default 100 ms)
0.1.52022-04-1912134Add PLAIN Auth
0.1.42022-02-1510186Add SCRAM-SHA-512 Auth
0.1.32022-02-1410256Add -XX:+ExitOnOutOfMemoryError JVM option
0.1.22021-12-218865Fix SASL config read issue
0.1.12021-12-068524Update connector fields title/description
Was this page helpful?