HTTP Source Connector

Poll REST APIs and stream responses to Kafka topics for integration with external services.

Overview

The HTTP Source Connector polls HTTP endpoints and writes responses to Kafka, supporting:

Scheduled polling at configurable intervals
Multiple authentication methods
Pagination handling
Response parsing and transformation

Installation

# Confluent Hub
confluent-hub install confluentinc/kafka-connect-http:latest

# Manual installation
curl -O https://packages.confluent.io/archive/7.5/kafka-connect-http-7.5.0.zip
unzip kafka-connect-http-7.5.0.zip -d /usr/share/kafka/plugins/

Basic Configuration

{
  "name": "http-source",
  "config": {
    "connector.class": "io.confluent.connect.http.HttpSourceConnector",
    "tasks.max": "1",

    "http.url": "https://api.example.com/events",
    "http.method": "GET",
    "kafka.topic": "api-events",

    "http.timer.interval.ms": "60000"
  }
}

HTTP Settings

Request Configuration

{
  "http.url": "https://api.example.com/data",
  "http.method": "GET",
  "http.headers": "Accept: application/json, X-API-Version: v2"
}

Property	Description	Default
`http.url`	Endpoint URL	Required
`http.method`	HTTP method	GET
`http.headers`	Request headers	None
`http.request.body`	Request body (POST/PUT)	None

POST Request

{
  "http.url": "https://api.example.com/query",
  "http.method": "POST",
  "http.headers": "Content-Type: application/json",
  "http.request.body": "{\"query\": \"SELECT * FROM events\"}"
}

Authentication

API Key

{
  "http.headers": "Authorization: Bearer ${secrets:api/token}"
}

Basic Authentication

{
  "http.auth.type": "BASIC",
  "http.auth.user": "${secrets:api/username}",
  "http.auth.password": "${secrets:api/password}"
}

OAuth 2.0

{
  "http.auth.type": "OAUTH2",
  "http.oauth2.token.url": "https://auth.example.com/oauth/token",
  "http.oauth2.client.id": "${secrets:oauth/client-id}",
  "http.oauth2.client.secret": "${secrets:oauth/client-secret}",
  "http.oauth2.scope": "read:events"
}

Polling Configuration

Fixed Interval

{
  "http.timer.interval.ms": "60000"
}

Poll every 60 seconds.

Cron Schedule

{
  "http.timer.type": "cron",
  "http.timer.cron.expression": "0 */5 * * * ?"
}

Poll every 5 minutes.

Offset Tracking

Track last processed item for incremental polling:

{
  "http.offset.mode": "TIMESTAMP",
  "http.offset.field": "updated_at",
  "http.initial.offset": "2024-01-01T00:00:00Z"
}

Response Handling

JSON Response

{
  "http.response.format": "JSON",
  "http.response.records.path": "$.data[*]"
}

Extract records from JSON array at $.data.

Array Response

{
  "http.response.format": "JSON",
  "http.response.records.path": "$[*]"
}

Process top-level array elements.

Single Record

{
  "http.response.format": "JSON",
  "http.response.records.path": "$"
}

Treat entire response as single record.

Pagination

Offset-Based

{
  "http.pagination.type": "OFFSET",
  "http.pagination.offset.param": "offset",
  "http.pagination.limit.param": "limit",
  "http.pagination.limit.value": "100"
}

URL: https://api.example.com/events?offset=0&limit=100

Cursor-Based

{
  "http.pagination.type": "CURSOR",
  "http.pagination.cursor.param": "cursor",
  "http.pagination.cursor.path": "$.next_cursor"
}

Extract cursor from response for next request.

Link Header

{
  "http.pagination.type": "LINK_HEADER"
}

Follow Link header for pagination (GitHub API style).

Error Handling

Retry Configuration

{
  "http.retry.max.attempts": "5",
  "http.retry.backoff.ms": "1000",
  "http.retry.backoff.max.ms": "60000"
}

Error Responses

{
  "http.response.error.codes": "400,401,403,404,500,502,503",
  "behavior.on.error": "fail"
}

Behavior	Description
`fail`	Stop connector
`log`	Log and continue
`ignore`	Silently skip

SSL/TLS

{
  "http.ssl.enabled": "true",
  "http.ssl.truststore.location": "/path/to/truststore.jks",
  "http.ssl.truststore.password": "${secrets:ssl/truststore-password}",
  "http.ssl.keystore.location": "/path/to/keystore.jks",
  "http.ssl.keystore.password": "${secrets:ssl/keystore-password}"
}

Complete Example

{
  "name": "github-events-source",
  "config": {
    "connector.class": "io.confluent.connect.http.HttpSourceConnector",
    "tasks.max": "1",

    "http.url": "https://api.github.com/repos/apache/kafka/events",
    "http.method": "GET",
    "http.headers": "Accept: application/vnd.github+json, Authorization: Bearer ${secrets:github/token}",

    "kafka.topic": "github-events",

    "http.timer.interval.ms": "300000",

    "http.response.format": "JSON",
    "http.response.records.path": "$[*]",

    "http.pagination.type": "LINK_HEADER",

    "http.retry.max.attempts": "3",
    "http.retry.backoff.ms": "5000",

    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",

    "errors.tolerance": "all",
    "errors.log.enable": "true"
  }
}

Transforms

Add Metadata

{
  "transforms": "addSource,addTimestamp",
  "transforms.addSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addSource.static.field": "source",
  "transforms.addSource.static.value": "github-api",
  "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTimestamp.timestamp.field": "ingested_at"
}

Extract Key

{
  "transforms": "extractKey",
  "transforms.extractKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
  "transforms.extractKey.fields": "id"
}

Rate Limiting

Throttling

{
  "http.timer.interval.ms": "60000",
  "http.throttle.max.requests": "100",
  "http.throttle.window.ms": "60000"
}

Maximum 100 requests per minute.

Respect Rate Limit Headers

{
  "http.rate.limit.header": "X-RateLimit-Remaining",
  "http.rate.limit.reset.header": "X-RateLimit-Reset"
}

Monitoring

Connector Status

curl http://connect:8083/connectors/http-source/status

Metrics

Metric	Description
`http-source-requests-total`	Total HTTP requests
`http-source-errors-total`	Failed requests
`http-source-records-polled`	Records retrieved

Troubleshooting

Issue	Cause	Solution
Connection timeout	Network/firewall	Verify endpoint accessibility
401 Unauthorized	Invalid credentials	Check authentication config
429 Too Many Requests	Rate limited	Increase polling interval
Empty response	Wrong JSON path	Verify `records.path`
Duplicate records	No offset tracking	Enable offset mode

Connectors Overview - All connectors
Kafka Connect - Connect framework
Transforms - Single Message Transforms