60-Day Kafka 4 Learning Plan · Week 6 — Day 38 of 60

60-Day Kafka 4 Learning Plan · Week 6 — Kafka Connect & ksqlDB Sources: Kafka: The Definitive Guide Ch.9 · docs.confluent.io/kafka-connectors/http-sink

Goal

Bridge Kafka topics to REST APIs without writing a single consumer. The HTTP Sink connector reads Kafka records and POSTs them to any HTTP endpoint — perfect for webhooks, microservice triggers, and third-party integrations — and understand the delivery-duplication, ordering, and security gaps a naive webhook receiver needs to handle.

1. What HTTP Sink does

labs.events (Kafka topic)
  → HTTP Sink Connector
  → POST /api/webhook/orders
  → labs-socket / 3rd-party REST endpoint

Use cases:

Push order events to labs-socket to trigger WebSocket broadcasts
Forward confirmed orders to a payment gateway webhook
Notify an external notification service without coupling it to Kafka
Trigger CI/CD pipelines from Kafka events

2. Basic connector config

Install the plugin

docker exec connect confluent-hub install confluentinc/kafka-connect-http:1.7.3 --no-prompt
docker-compose restart connect

Deploy the connector

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-http-sink",
    "config": {
      "connector.class": "io.confluent.connect.http.HttpSinkConnector",
      "topics": "labs.events",
      "http.api.url": "http://labs-socket:8090/api/webhook/orders",
      "request.method": "POST",
      "headers": "Content-Type:application/json,X-Source:kafka",
      "batch.max.size": "10",
      "tasks.max": "2",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter.schemas.enable": "false"
    }
  }'

Key config properties

Dynamic URL routing

Route different topics to different endpoints using URL patterns:

"http.api.url": "http://labs-socket:8090/api/webhook/${topic}",
"topics": "labs.events,labs.notifications,labs.alerts"

labs.events → POST /api/webhook/labs.events
labs.notifications → POST /api/webhook/labs.notifications

3. Retry & error handling

HTTP endpoints are unreliable. Always configure retries and a dead-letter queue (DLQ).

# Add these properties to the config block above
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "labs-http-sink-prod",
    "config": {
      "connector.class": "io.confluent.connect.http.HttpSinkConnector",
      "topics": "labs.events",
      "http.api.url": "http://labs-socket:8090/api/webhook/orders",
      "request.method": "POST",
      "headers": "Content-Type:application/json",
      "batch.max.size": "10",
      "tasks.max": "2",

      "retry.backoff.ms": "1000",
      "max.retries": "5",
      "errors.retry.timeout": "30000",

      "errors.tolerance": "all",
      "errors.deadletterqueue.topic.name": "labs.dlq.http-sink",
      "errors.deadletterqueue.topic.replication.factor": "1",
      "errors.deadletterqueue.context.headers.enable": "true",
      "errors.log.enable": "true",
      "errors.log.include.messages": "true"
    }
  }'

Same dev-only setting flagged before, worth catching again here: errors.deadletterqueue.topic.replication.factor: 1 — exactly the pattern called out for the internal Connect topics in Day 36 §4 and the Debezium DLQ example in Day 37, just showing up a third time in this connector’s own DLQ. For a topic whose entire purpose is preserving failed records for later investigation, RF=1 means a single broker failure silently loses the evidence of what went wrong — set this to 3 in any environment where the DLQ’s contents actually matter.

What goes to the DLQ?

Records that hit 4xx responses (client errors — not retried)
Records that exhaust all retries after 5xx / network errors
Records that fail to deserialize

Each DLQ record includes error context headers explaining the failure reason.

4. Authentication options

No auth (internal services)

"auth.type": "NONE"

Safe for internal service-to-service calls on a private network (e.g. labs-socket on the same Docker network).

Bearer token (static API key)

"headers": "Content-Type:application/json,Authorization:Bearer ${file:/secrets/token.txt:api_key}"

Use an external secrets file to avoid putting credentials in config.

OAuth 2.0 ★ (auto-refreshing)

"auth.type": "OAUTH2",
"oauth2.token.url": "https://auth.example.com/oauth/token",
"oauth2.client.id": "labs-connect-client",
"oauth2.client.secret": "${file:/secrets/oauth.txt:client_secret}",
"oauth2.token.property": "access_token"

The connector automatically refreshes the token before it expires — no manual rotation needed.

5. SMT — transform payload before posting

Single Message Transforms (SMTs) reshape records in-flight between Kafka and the HTTP endpoint.

Drop unwanted fields (ReplaceField)

"transforms": "extractFields",
"transforms.extractFields.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.extractFields.whitelist": "orderId,userId,amount,status"

Only orderId, userId, amount, and status are included in the POST body.

Add static metadata (InsertField)

"transforms": "addSource,addTimestamp",
"transforms.addSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.addSource.static.field": "source",
"transforms.addSource.static.value": "kafka-connect",
"transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.addTimestamp.timestamp.field": "processedAt"

Rename fields (ReplaceField)

"transforms": "renameFields",
"transforms.renameFields.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
"transforms.renameFields.renames": "orderId:id,userId:user_id"

Chain multiple SMTs

"transforms": "extractFields,addSource,renameFields"

SMTs execute left-to-right. Each transform receives the output of the previous one.

6. HTTP response code handling

Monitoring failed deliveries

# Consume from DLQ to inspect failed records
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic labs.dlq.http-sink \
  --from-beginning \
  --property print.headers=true

7. Receiving the webhook in labs-socket (Spring Boot)

// WebhookController.java — labs-socket receiving Kafka events via HTTP Sink
@RestController
@RequiredArgsConstructor
public class WebhookController {

    private final SimpMessagingTemplate websocket;

    // Receives a single event (batch.max.size = 1)
    @PostMapping("/api/webhook/orders")
    public ResponseEntity<Void> handleOrder(@RequestBody OrderEvent event,
                                            @RequestHeader("X-Source") String source) {
        log.info("Received from {}: order={}", source, event.getOrderId());

        // Broadcast to WebSocket subscribers
        websocket.convertAndSend("/topic/orders", event);

        return ResponseEntity.ok().build();   // 200 → connector commits offset
    }

    // Receives a batch (batch.max.size > 1)
    @PostMapping("/api/webhook/orders/batch")
    public ResponseEntity<Void> handleBatch(@RequestBody List<OrderEvent> events) {
        events.forEach(e -> websocket.convertAndSend("/topic/orders", e));
        return ResponseEntity.ok().build();
    }
}

8. At-least-once delivery — retries can duplicate POSTs

The HTTP Sink connector’s retry-on-5xx/timeout behavior (§6) is a form of at-least-once delivery, and it carries the exact same duplication risk covered for Kafka producers in Day 16 §3 — just at the HTTP layer instead of the Kafka protocol layer.

Connector POSTs event → labs-socket processes it successfully → broadcasts to WebSocket
                       → response is lost in transit (network blip) before connector sees 200
Connector times out → retries → labs-socket receives the SAME event again
                                → broadcasts it AGAIN → duplicate WebSocket message

The webhook receiver in §7 has no protection against this — it processes and broadcasts every request it receives, with no way to recognize “I’ve already handled this exact event.” For a WebSocket broadcast, a duplicate might just be a minor UX glitch (the same notification flashing twice); for anything triggering a side effect with real consequences (a payment webhook, an inventory decrement), it’s a correctness bug.

@PostMapping("/api/webhook/orders")
public ResponseEntity<Void> handleOrder(@RequestBody OrderEvent event,
                                        @RequestHeader("X-Source") String source) {
    // Idempotency check — dedupe by the event's own ID, not a connector-generated one
    if (!processedEventIds.putIfAbsent(event.getOrderId(), Instant.now())) {
        log.info("Duplicate webhook delivery for order={}, skipping", event.getOrderId());
        return ResponseEntity.ok().build(); // still 200 — tell the connector it's "done"
    }

    websocket.convertAndSend("/topic/orders", event);
    return ResponseEntity.ok().build();
}

Use a real deduplication store for this in production (Redis with a TTL, or a DB unique constraint on orderId), not an unbounded in-memory map as shown — the map above is illustrative only and would leak memory over time. The core principle: the receiver, not the connector, is responsible for idempotency — HTTP Sink’s job is “deliver each Kafka record at least once,” and turning that into “handled effectively once” is the receiving application’s job, exactly analogous to how Kafka’s own idempotent producer (Day 16) solves duplication at the broker layer but a downstream consumer still needs its own idempotency strategy for anything beyond simple offset tracking.

9. Ordering vs parallelism — the `tasks.max` trade-off

tasks.max: 2 (§2’s example) means two connector tasks run concurrently, each handling a subset of the source topic’s partitions — the same parallelism model as any Connect deployment (Day 36 §3). This has a direct consequence for delivery order that’s easy to overlook when just copying a config.

Within a single partition, HTTP Sink delivers records in order (subject to retries — see below).
Across partitions handled by different tasks, there’s no ordering guarantee between them — two events for different keys can be POSTed in any relative order, arriving at the webhook receiver out of sequence relative to when they were produced.
A retry on one task doesn’t block other tasks — if task 1 is retrying a failed POST, task 2 keeps delivering its own partition’s records normally, meaning a slow/failing endpoint interaction on one partition doesn’t stall the whole pipeline, but does mean “global” ordering across the topic was never actually guaranteed even before any retry happened.

Practical implication: if the receiving webhook needs to process events for a given key (e.g. orderId) in the order they happened, that’s only guaranteed if the source topic is partitioned by that key (Day 12) — tasks.max parallelism doesn’t break per-key ordering, but it does mean you shouldn’t assume any ordering guarantee across different keys, and batch.max.size > 1 similarly only guarantees relative order within a single batched POST, not across separate HTTP requests from different tasks.

10. Securing the webhook receiver

The X-Source: kafka header in §2/§7 is trivially spoofable — any client that knows (or guesses) the endpoint URL can send a POST with that same header and have it treated as a legitimate Kafka-originated event. It’s metadata for logging, not an authentication mechanism, even though it might look like one at a glance.

@PostMapping("/api/webhook/orders")
public ResponseEntity<Void> handleOrder(@RequestBody String rawBody,
                                        @RequestHeader("X-Signature") String signature) {
    String expectedSignature = hmacSha256(rawBody, webhookSharedSecret);
    if (!MessageDigest.isEqual(signature.getBytes(), expectedSignature.getBytes())) {
        return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
    }
    // ... proceed to parse and process rawBody as OrderEvent
}

"transforms": "addSignature",
"headers": "Content-Type:application/json,X-Signature:${hmac}"

The general principle, same as Day 36 §11’s REST API and Day 37 §8’s connector credentials: any HTTP endpoint receiving data that drives real behavior (WebSocket broadcasts, downstream side effects) needs to verify the request actually came from the expected source — a header the connector sets is only as trustworthy as the network path is private. For genuinely internal, network-isolated services (labs-socket reachable only from the Connect cluster’s own Docker network) this is lower priority; for anything reachable from a broader network or the public internet, signature verification (HMAC with a shared secret, matching the OAuth2 pattern from §4 for outbound auth) is the minimum bar.

11. Common pitfalls

Leaving the DLQ topic at replication.factor=1 — the third time this exact pattern shows up (Day 36 §4, Day 37 examples, now here) — worth internalizing as a default habit to check, not just a one-off fix (§3)
Assuming the webhook receiver is naturally idempotent — at-least-once HTTP delivery means retries can and will duplicate POSTs; the receiver must dedupe explicitly (§8)
Assuming tasks.max > 1 preserves global ordering across the whole topic — it only preserves per-partition order, same as any other Kafka parallelism (§9)
Treating X-Source or similar plain headers as authentication — they’re spoofable metadata, not a security control; use signature verification for anything beyond a fully isolated internal network (§10)
Not planning for the webhook receiver being temporarily down — without errors.tolerance: all and appropriate max.retries/backoff tuning, a brief receiver outage can either stall the whole connector task or silently drop records depending on configuration; test this failure mode deliberately rather than discovering the behavior during a real outage

Key Takeaways

HTTP Sink bridges Kafka → REST with no custom consumer code — just JSON config
batch.max.size batches multiple records into one POST body (array payload)
5xx errors are retried with backoff; 4xx are fatal — sent to DLQ immediately
SMT (Single Message Transform) reshapes records before they are POSTed
OAuth 2.0 auth type auto-refreshes tokens — no manual token rotation needed
Retries mean at-least-once delivery to the webhook — the receiver, not the connector, is responsible for idempotency if duplicates matter
tasks.max parallelism preserves per-partition order only — never assume ordering across different keys/partitions
A plain header like X-Source is spoofable metadata, not authentication — use signature verification for endpoints reachable beyond a fully trusted network
Use errors.tolerance: all + a properly-replicated DLQ to never block the pipeline on bad records

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

➡️ Day 39: Redis Pub/Sub sink — Kafka → Redis for labs-socket

Resources

📘 Kafka: The Definitive Guide — Chapter 9 (Kafka Connect)
🌐 docs.confluent.io/kafka-connectors/http-sink

Link to Medium blog