Bright Data Redpanda Connect Plugin

A Redpanda Connect Dynamic Plugin that exposes Bright Data's web intelligence APIs — Web Unlocker, SERP, Web Scraper, and Dataset — as native pipeline processor components. Messages flow through your Redpanda Connect pipelines and are enriched with live web data from Bright Data without requiring any custom HTTP boilerplate.

---

Prerequisites

• Python 3.11 or higher
• pip
• Redpanda Connect v4.56.0 or higher (rpk connect)
• Docker (optional, for containerised deployment)
• A valid Bright Data account and API key

---

Installation

1. Clone the repository and enter the project directory:

   git clone <repo-url>
   cd brightdata-redpanda-plugin

2. Create and activate a virtual environment:

   python -m venv venv
   source venv/bin/activate   # Windows: venv\Scripts\activate

3. Install dependencies:

   pip install -r requirements.txt

4. Copy the environment template and fill in your credentials:

   cp .env.example .env
   # Edit .env with your BRIGHTDATA_API_KEY and zone names

---

Environment Variables

Variable	Required	Default	Description
BRIGHTDATA_API_KEY	Yes	—	Your Bright Data API key
BRIGHTDATA_UNLOCKER_ZONE	No	web_unlocker1	Zone name for Web Unlocker operations
BRIGHTDATA_SERP_ZONE	No	serp_api1	Zone name for SERP operations
BRIGHTDATA_COUNTRY	No	us	Default country code for geo-targeting
BRIGHTDATA_POLL_INTERVAL	No	10	Seconds between async job poll requests
LOG_LEVEL	No	INFO	Python logging level (DEBUG, INFO, WARNING, ERROR)

---

Running Locally

Start a pipeline using rpk connect run with the plugin manifest:

rpk connect run --rpc-plugins=plugin.yaml pipeline.yaml

Send a test message:

echo '{"url":"https://example.com"}' | \
  rpk connect run --rpc-plugins=plugin.yaml examples/web_unlocker_pipeline.yaml

---

Configuration Reference

Shared Parameters (all operations)

Parameter	Type	Default	Description
api_key	string	$BRIGHTDATA_API_KEY	Bright Data API key
operation	enum	—	web_unlocker, serp, scraper, or dataset
zone	string	""	Zone name (falls back to env var)
timeout_seconds	int	30	Per-request HTTP timeout
retry_max	int	3	Maximum retry attempts on retryable errors
retry_backoff_ms	int	500	Base back-off delay in milliseconds

Web Unlocker Parameters

Parameter	Type	Default	Description
url_field	string	url	Message field containing the target URL
render_js	bool	false	Enable JavaScript rendering
country	string	us	Geo-targeting country code

SERP Parameters

Parameter	Type	Default	Description
query_field	string	query	Message field containing the search query
engine	enum	google	google or bing
hl	string	en	Results language code
gl	string	us	Results country code
parsed_json	bool	true	Request structured JSON (brd_json=1)

Web Scraper Parameters

Parameter	Type	Default	Description
dataset_id	string	""	Bright Data dataset ID
input_field	string	urls	Message field containing the URL array
async_mode	bool	true	Use async trigger/poll/retrieve flow
poll_interval_seconds	int	10	Polling interval for async jobs
poll_timeout_seconds	int	300	Maximum wait time for async jobs

Dataset Parameters

Parameter	Type	Default	Description
dataset_id	string	""	Bright Data dataset ID to snapshot
poll_interval_seconds	int	10	Polling interval
poll_timeout_seconds	int	300	Maximum wait time

---

Worked Examples

Web Unlocker

pipeline:
  processors:
    - brightdata:
        api_key: "${BRIGHTDATA_API_KEY}"
        operation: web_unlocker
        zone: web_unlocker1
        url_field: url
        render_js: false
        country: us
        timeout_seconds: 30
        retry_max: 3
        retry_backoff_ms: 500

Send: {"url": "https://example.com"}

SERP API

pipeline:
  processors:
    - brightdata:
        api_key: "${BRIGHTDATA_API_KEY}"
        operation: serp
        zone: serp_api1
        query_field: query
        engine: google
        hl: en
        gl: us
        parsed_json: true

Send: {"query": "redpanda connect streaming"}

Web Scraper (async)

pipeline:
  processors:
    - brightdata:
        api_key: "${BRIGHTDATA_API_KEY}"
        operation: scraper
        dataset_id: "gd_l1viktl72bvnsativi"
        input_field: urls
        async_mode: true
        poll_interval_seconds: 10
        poll_timeout_seconds: 300

Send: {"urls": [{"url": "https://example.com"}, {"url": "https://redpanda.com"}]}

Dataset API

pipeline:
  processors:
    - brightdata:
        api_key: "${BRIGHTDATA_API_KEY}"
        operation: dataset
        dataset_id: "gd_l1viktl72bvnsativi"
        poll_interval_seconds: 10
        poll_timeout_seconds: 300

Send: {}

---

Running Tests

Run all unit tests:

pytest tests/unit/ -v

Run integration tests (no live API calls — all HTTP is mocked):

pytest tests/integration/ -v

Run the full suite:

pytest -v

---

Building the Docker Image

docker build -t brightdata-redpanda-plugin:latest .

Run the containerised plugin alongside rpk connect:

docker run --rm \
  -e BRIGHTDATA_API_KEY=your_api_key_here \
  -e BRIGHTDATA_UNLOCKER_ZONE=web_unlocker1 \
  brightdata-redpanda-plugin:latest

---

Error Handling

Every failed message is nacked with a structured error annotation attached to metadata.brightdata_error:

{
  "error_code": "HTTP_429",
  "message": "Too Many Requests",
  "retryable": true,
  "attempt": 2,
  "request_id": "req-abc123",
  "timestamp": "2026-06-04T10:00:00+00:00"
}

HTTP Code	Behaviour
200	Success — message enriched with brightdata key
400	Nacked immediately, not retried
401 / 403	Pipeline halted — check credentials
404	Nacked, not retried
429	Retried with exponential back-off (respects Retry-After)
500 / 502 / 503	Retried with exponential back-off
Timeout	Retried up to retry_max, then nacked