Skip to main content

​
Overview

The Twitter Source Connector uses Twitter4J to connect to the legacy Twitter Streaming API. It continuously receives real-time statuses that match configured keywords and optional user IDs, converts each status into a structured Kafka Connect record, and writes the record to a specified Kafka topic. It is intended for ingesting a real-time status stream that the account is authorized to access into Kafka for stream processing, search, or archival systems. Regular statuses are written to kafka.status.topic. When delete-notice processing is enabled, the connector also writes delete notices to the same topic as records with null values. The connector processes only real-time callbacks received after the connection is established; it does not provide historical search, backfill existing data, or maintain a recoverable upstream read position.

​
Prerequisites

  • Kafka Connect 3.9.1 cannot use the official 0.3.34 ZIP directly. The release package is missing the Guava dependency required at runtime. Even after adding the Guava 30.1.1-jre version declared by its dependency management, the packaged connect-utils 0.7.171 still throws an IllegalAccessError when accessing SystemTime in Kafka 3.9.1, preventing the task from starting. Proceed only after obtaining an official artifact that the connector vendor provides and confirms as compatible with Kafka Connect 3.9.1.
  • The connector uses Twitter4J 4.0.6 to access the legacy statuses/filter.json streaming endpoint. Before deployment, confirm with the current X/Twitter API service that this endpoint remains available to the target account, product tier, and region. Existing OAuth credentials alone do not establish availability.
  • Prepare an OAuth 1.0a consumer key, consumer secret, access token, and access token secret authorized to access that streaming endpoint, and confirm that the authorization scope permits reading the required real-time data.
  • Identify the keywords to track. filter.keywords is required; configuring only user IDs does not create an ingestion task.

​
License

Licensed under the Apache License 2.0.

​
Quick Start

The configuration below is for reference only and does not establish that the official 0.3.34 ZIP can run on Kafka Connect 3.9.1. First obtain an official artifact that the vendor has confirmed as compatible with Kafka Connect 3.9.1; do not add dependencies or modify binaries yourself to bypass the compatibility errors. Then prepare a Connect cluster, Kafka, a topic to receive status records, and an X/Twitter account with OAuth credentials whose access to the legacy streaming endpoint has been confirmed. Also verify network connectivity and access permissions. For preparation and management procedures, see Manage Connectors.
Replace the keyword, target topic, and four OAuth placeholders. Supply credentials through a controlled configuration-management mechanism; do not commit them to a source repository or write them to logs. The example inherits the worker’s key and value converters. The selected converters must be able to serialize nested Structs, arrays, maps, and Timestamps. Before submitting the configuration, confirm with the X/Twitter API service that the target account currently has permission to use the legacy streaming endpoint; successful configuration validation does not establish that the external real-time stream is available.

​
Configuration

​
Authentication

​
twitter.oauth.consumerKey

The Twitter OAuth consumer key.
  • Type: password
  • Default: None
  • Importance: High
  • Valid Values / Notes: Use a value that belongs to the same application and authorization relationship as the other three OAuth settings. Credential validity is checked by the external API.
  • Required: Yes

​
twitter.oauth.consumerSecret

The Twitter OAuth consumer secret.
  • Type: password
  • Default: None
  • Importance: High
  • Valid Values / Notes: Manage this as sensitive configuration. Do not write it to logs, support tickets, or shared examples.
  • Required: Yes

​
twitter.oauth.accessToken

The Twitter OAuth access token.
  • Type: password
  • Default: None
  • Importance: High
  • Valid Values / Notes: It must correspond to an account and application authorized to access the legacy streaming endpoint. Connector configuration validation does not confirm actual permissions.
  • Required: Yes

​
twitter.oauth.accessTokenSecret

The Twitter OAuth access token secret.
  • Type: password
  • Default: None
  • Importance: High
  • Valid Values / Notes: Manage this as sensitive configuration and use it with the corresponding access token.
  • Required: Yes

​
Ingestion Scope

​
filter.keywords

The list of tracking keywords submitted to the streaming filter endpoint.
  • Type: list
  • Default: None
  • Importance: High
  • Valid Values / Notes: Separate keywords with commas. Provide at least one valid keyword. The connector partitions tasks according to the number of keywords and tasks.max.
  • Required: Yes

​
filter.userIds

The list of user IDs to track through the streaming filter endpoint.
  • Type: list
  • Default: Empty list
  • Importance: High
  • Valid Values / Notes: Use comma-separated decimal numeric user IDs, not usernames. Each item must be parseable as a Java long. Duplicate IDs are merged. This setting does not replace the required filter.keywords setting.

​
Kafka Destination and Delete Notices

​
kafka.status.topic

The Kafka topic that receives regular statuses and, when enabled, delete notices.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: Specify a single explicit topic name. This version does not support a separate kafka.delete.topic.
  • Required: Yes

​
process.deletes

Controls whether Twitter delete notices are processed.
  • Type: boolean
  • Default: None
  • Importance: High
  • Valid Values / Notes: false ignores delete notices. true writes each delete notice to kafka.status.topic with a StatusDeletionNoticeKey containing only StatusId and a null value.
  • Required: Yes

​
Queue and Batching

​
queue.empty.ms

The time, in milliseconds, to wait for records when the internal queue is empty.
  • Type: int
  • Default: 100
  • Importance: Low
  • Valid Values / Notes: The minimum is 10. This setting controls how long an empty-queue poll waits; it is not a Twitter API connection timeout.

​
queue.batch.size

The target number of records returned from the internal queue by one poll.
  • Type: int
  • Default: 100
  • Importance: Low
  • Valid Values / Notes: The minimum is 1. This setting controls connector-internal queue batches; it is not a Twitter API fetch batch size.

​
Twitter Client Diagnostics

​
twitter.debug

Enables Twitter4J debug logging.
  • Type: boolean
  • Default: false
  • Importance: Low
  • Valid Values / Notes: Enable it temporarily and only during controlled troubleshooting. It may produce substantial output and does not change the Kafka Connect worker’s log-level configuration.

​
Connector Identity and Tasks

​
connector.class

The connector implementation class to load.
  • Type: string
  • Default: None
  • Importance: High
  • Valid Values / Notes: Use com.github.jcustenborder.kafka.connect.twitter.TwitterSourceConnector.
  • Required: Yes

​
tasks.max

The maximum number of tasks the connector may create.
  • Type: int
  • Default: 1
  • Importance: High
  • Valid Values / Notes: The minimum is 1. The actual task count is the smaller of tasks.max and the number of keywords. Increasing this value establishes more independent external streaming connections.

​
tasks.max.enforce

Controls whether Kafka Connect enforces the tasks.max limit.
  • Type: boolean
  • Default: true
  • Importance: Low
  • Valid Values / Notes: Kafka Connect 3.9.1 marks this setting as deprecated and plans to remove it. Keep it set to true and manage the task count through tasks.max.
  • Deprecation Status: Yes

​
Record Conversion

​
key.converter

Overrides the worker converter used to serialize SourceRecord keys.
  • Type: class
  • Default: null, inherits the worker configuration
  • Importance: Low
  • Valid Values / Notes: If explicitly configured, the class must implement the Kafka Connect Converter interface and be instantiable. Regular statuses and delete notices use structured keys.

​
value.converter

Overrides the worker converter used to serialize SourceRecord values.
  • Type: class
  • Default: null, inherits the worker configuration
  • Importance: Low
  • Valid Values / Notes: If explicitly configured, the class must implement the Kafka Connect Converter interface and be instantiable. Regular statuses contain nested structures, arrays, maps, and Timestamps; delete-notice values are null.

​
Best Practices

​
Adjust the Scope of Ongoing Ingestion

Applicable scenario: After the connector has completed initial onboarding, business topics change and you need to add or narrow keywords. You may also need to track a defined set of numeric user IDs so that subsequent real-time data continues to enter the existing target topic. Replace or add the following properties in the Quick Start configuration:
Key considerations: filter.userIds accepts only numeric IDs and cannot create tasks by itself, so always retain at least one keyword. Changing the filter conditions rebuilds the real-time stream. Because the connector has no historical backfill or checkpoint-resume capability, account for possible gaps or duplicates during the transition window in the downstream design.

​
Increase Ingestion Parallelism for Multiple Keywords

Applicable scenario: An existing single task has been running continuously, and an expanded keyword set now needs to be distributed across multiple independent tasks and streaming connections to reduce the filtering scope handled by each connection. Replace or add the following properties in the Quick Start configuration:
Key considerations: The connector never creates more tasks than there are keywords, and it assigns keywords in iteration order using round-robin distribution. Increasing tasks.max establishes more external streaming connections. If filter.userIds is also configured, the complete user ID list is copied to every task, which may produce duplicate records. The connector does not guarantee global ordering or deduplication across tasks; downstream systems should use stable business identifiers to handle duplicates.

​
Preserve Upstream Delete Signals

Applicable scenario: Continuous ingestion is already running, and downstream archival, indexing, or compliance processing needs to observe status delete notices emitted upstream instead of silently ignoring them. Replace the following property in the Quick Start configuration:
Key considerations: Delete notices are written to the same kafka.status.topic as regular statuses. Each record uses a StatusDeletionNoticeKey containing only StatusId and has a null value; this version has no separate delete topic. Downstream systems must explicitly recognize this record shape. If Kafka log compaction is expected to perform deletion, also confirm that the final key serialization, partitioning, and target-topic configuration match the intended downstream semantics.

​
Monitoring

​
What to Monitor

Monitor general Kafka Connect health, connector and task states, throughput, latency, offset commits, errors, retries, and worker JVM signals. Monitor DLQ activity only when the deployment has enabled the corresponding error handling. This connector does not persist a recoverable Twitter upstream cursor, so a task in the RUNNING state does not establish that the external real-time stream is complete. Also inspect logs related to authentication, connections, stall warnings, and upstream rate limits.

​
Import the Grafana Dashboard

Confirm that Connect metrics are available through a Grafana data source and that collected labels satisfy the dashboard filters. Download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

​
Limitations

  • The official 0.3.34 ZIP is missing Guava, so Kafka Connect 3.9.1 cannot complete plugin discovery. The unmodified release package cannot be used directly with this worker version.
  • After Guava 30.1.1-jre is added to 0.3.34, connect-utils 0.7.171 still accesses SystemTime in Kafka 3.9.1 when the task starts and throws an IllegalAccessError. This failure occurs before any X/Twitter request, and the task cannot run.
  • The connector depends on the legacy statuses/filter.json streaming endpoint used by Twitter4J 4.0.6. Its current availability, product-tier access, and account authorization are determined by the X/Twitter API service and must be confirmed by the user before deployment.
  • The connector receives only real-time statuses and delete callbacks after the connection is established. It does not support historical search, backfilling existing data, or reading a specified time range.
  • The connector does not persist an upstream cursor that can be used for recovery. After a task restart, migration, or filter-condition change, it establishes a new real-time connection and cannot resume precisely or backfill the interruption window.
  • The connector does not guarantee global ordering across tasks, idempotent writes, or deduplication. Multiple tasks, overlapping filter conditions, and reconnections can all introduce duplicates or gaps.
  • All regular statuses are written to one kafka.status.topic. The connector does not route records to separate topics by keyword, user, or event type, and it does not provide a raw JSON passthrough mode.
  • Delete notices can be written only as records with null values in kafka.status.topic. This version does not support kafka.delete.topic.
  • The internal queue is effectively unbounded under practical workloads and has no upstream-pause mechanism. If Kafka writes or serialization remain slow, the backlog may increase worker memory pressure.
  • Per-record conversion or enqueue exceptions caught internally by the connector are not passed to Kafka Connect’s DLQ handling, and the event is not retried automatically.

​
FAQ

​
Why Can’t the Official 0.3.34 ZIP Start a Task in Kafka Connect 3.9.1?

The original release package is missing Guava, so the worker reports that com.google.common.collect.Multimap is missing during plugin discovery. After only Guava 30.1.1-jre is added, the plugin can be discovered and accepts the configuration, but the task throws an IllegalAccessError because connect-utils 0.7.171 accesses SystemTime in Kafka 3.9.1. The error occurs before the X/Twitter client is created or any external request is sent, so it is unrelated to OAuth credentials or external API responses. Do not treat adding dependencies yourself or modifying the binary as a supported solution. Until the vendor provides and confirms an official artifact compatible with Kafka Connect 3.9.1, the examples on this page are configuration references only.

​
The Connector Cannot Establish a Twitter Streaming Connection

Common causes include the legacy streaming endpoint no longer being available to the target account, the four OAuth 1.0a credentials not belonging to the same authorization relationship, insufficient token permissions, or the external service rejecting the current product tier. First confirm the current availability of statuses/filter.json and the account authorization with the X/Twitter API service. Then verify the four credentials and the worker’s access path to the external service. Do not troubleshoot by printing complete credentials in logs.

​
The Connector Has Started but Has No Task or Produces No Records

If filter.keywords is empty, configuring only filter.userIds does not create an ingestion task. Confirm that at least one valid keyword is present, inspect connector and task states and authentication logs, and verify that the keywords or numeric user IDs satisfy the current streaming endpoint’s server-side rules. The streaming endpoint returns only matching events received after the connection is established; it does not backfill earlier data.

​
Duplicate Records or Ordering Changes Appear After Scaling Out

Each task uses an independent streaming connection. Keywords are partitioned, while the complete filter.userIds list is copied to every task. The same event may be duplicated because filter scopes overlap or connections are re-established, and there is no global ordering guarantee across tasks. Check tasks.max, keyword grouping, and user ID settings. Deduplicate downstream using stable status IDs; do not treat Kafka offsets as Twitter upstream cursors.

​
Why Is There No Separate Delete Topic After Delete Processing Is Enabled?

This version has no kafka.delete.topic setting. When process.deletes=true, delete notices are written to kafka.status.topic as records with delete keys and null values. Check whether downstream converters and consumers preserve and recognize null values, and confirm that the target topic’s compaction policy matches the intended behavior.