Kafka Upstream: Forward HTTP headers as Kafka record headers (allowlist mode) - Plugin

Forward HTTP headers as Kafka record headers (allowlist mode)v3.15+

Use config.headers to copy HTTP request headers onto the produced Kafka record as native record headers. This makes routing, tracing, and tenancy context available to consumers without parsing the message payload.

In allowlist mode (forward_all_by_default: false), only headers listed in include_headers are forwarded. Use name_mappings to rename an HTTP header to a different record header key, for example renaming x-request-id to trace-id.

Prerequisites

Kafka installed and running
Create a Kafka topic

Environment variables

KAFKA_TOPIC: The name of your Kafka topic.

Set up the plugin

Add this section to your kong.yaml configuration file:

kong.yaml

Copied!

_format_version: "3.0"
plugins:
  - name: kafka-upstream
    config:
      topic: ${{ env "DECK_KAFKA_TOPIC" }}
      headers:
        forward_http_headers_as_record_headers: true
        forward_all_by_default: false
        include_headers:
        - x-request-id
        - x-tenant-id
        - x-correlation-id
        name_mappings:
          x-request-id: trace-id
          x-tenant-id: tenant
        repeated_headers_behavior: concatenate_by_comma

Make the following request:

curl -i -X POST http://localhost:8001/plugins/ \
    --header "Accept: application/json" \
    --header "Content-Type: application/json" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make the following request:

curl -X POST https://{region}.api.konghq.com/v2/control-planes/{controlPlaneId}/core-entities/plugins/ \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

region: Geographic region where your Kong Konnect is hosted and operates.
KONNECT_TOKEN: Your Personal Access Token (PAT) associated with your Konnect account.
controlPlaneId: The id of the control plane.

See the Konnect API reference to learn about region-specific URLs and personal access tokens.

echo "
apiVersion: configuration.konghq.com/v1
kind: KongClusterPlugin
metadata:
  name: kafka-upstream
  namespace: kong
  annotations:
    kubernetes.io/ingress.class: kong
    konghq.com/tags: ''
  labels:
    global: 'true'
config:
  topic: '$KAFKA_TOPIC'
  headers:
    forward_http_headers_as_record_headers: true
    forward_all_by_default: false
    include_headers:
    - x-request-id
    - x-tenant-id
    - x-correlation-id
    name_mappings:
      x-request-id: trace-id
      x-tenant-id: tenant
    repeated_headers_behavior: concatenate_by_comma
plugin: kafka-upstream
" | kubectl apply -f -

Copied!

Prerequisite: Configure your Personal Access Token

terraform {
  required_providers {
    konnect = {
      source  = "kong/konnect"
    }
  }
}

provider "konnect" {
  personal_access_token = "$KONNECT_TOKEN"
  server_url            = "https://us.api.konghq.com/"
}

Copied!

Add the following to your Terraform configuration to create a Konnect Gateway Plugin:

resource "konnect_gateway_plugin_kafka_upstream" "my_kafka_upstream" {
  enabled = true

  config = {
    topic = var.kafka_topic

    headers = {
      forward_http_headers_as_record_headers = true
      forward_all_by_default = false
      include_headers = ["x-request-id", "x-tenant-id", "x-correlation-id"]

      name_mappings = {
        x-request-id = "trace-id"
        x-tenant-id = "tenant"
      }
      repeated_headers_behavior = "concatenate_by_comma"
    }
  }
  tags = []

  control_plane_id = konnect_gateway_control_plane.my_konnect_cp.id
}

Copied!

This example requires the following variables to be added to your manifest. You can specify values at runtime by setting TF_VAR_name=value.

variable "kafka_topic" {
  type = string
}

Copied!

Add this section to your kong.yaml configuration file:

kong.yaml

Copied!

_format_version: "3.0"
plugins:
  - name: kafka-upstream
    service: serviceName|Id
    config:
      topic: ${{ env "DECK_KAFKA_TOPIC" }}
      headers:
        forward_http_headers_as_record_headers: true
        forward_all_by_default: false
        include_headers:
        - x-request-id
        - x-tenant-id
        - x-correlation-id
        name_mappings:
          x-request-id: trace-id
          x-tenant-id: tenant
        repeated_headers_behavior: concatenate_by_comma

Make sure to replace the following placeholders with your own values:

serviceName|Id: The id or name of the service the plugin configuration will target.

Make the following request:

curl -i -X POST http://localhost:8001/services/{serviceName|Id}/plugins/ \
    --header "Accept: application/json" \
    --header "Content-Type: application/json" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

serviceName|Id: The id or name of the service the plugin configuration will target.

Make the following request:

curl -X POST https://{region}.api.konghq.com/v2/control-planes/{controlPlaneId}/core-entities/services/{serviceId}/plugins/ \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

region: Geographic region where your Kong Konnect is hosted and operates.
KONNECT_TOKEN: Your Personal Access Token (PAT) associated with your Konnect account.
controlPlaneId: The id of the control plane.
serviceId: The id of the service the plugin configuration will target.

See the Konnect API reference to learn about region-specific URLs and personal access tokens.

echo "
apiVersion: configuration.konghq.com/v1
kind: KongPlugin
metadata:
  name: kafka-upstream
  namespace: kong
  annotations:
    kubernetes.io/ingress.class: kong
    konghq.com/tags: ''
config:
  topic: '$KAFKA_TOPIC'
  headers:
    forward_http_headers_as_record_headers: true
    forward_all_by_default: false
    include_headers:
    - x-request-id
    - x-tenant-id
    - x-correlation-id
    name_mappings:
      x-request-id: trace-id
      x-tenant-id: tenant
    repeated_headers_behavior: concatenate_by_comma
plugin: kafka-upstream
" | kubectl apply -f -

Copied!

Next, apply the KongPlugin resource by annotating the service resource:

kubectl annotate -n kong service SERVICE_NAME konghq.com/plugins=kafka-upstream

Copied!

Prerequisite: Configure your Personal Access Token

terraform {
  required_providers {
    konnect = {
      source  = "kong/konnect"
    }
  }
}

provider "konnect" {
  personal_access_token = "$KONNECT_TOKEN"
  server_url            = "https://us.api.konghq.com/"
}

Copied!

Add the following to your Terraform configuration to create a Konnect Gateway Plugin:

resource "konnect_gateway_plugin_kafka_upstream" "my_kafka_upstream" {
  enabled = true

  config = {
    topic = var.kafka_topic

    headers = {
      forward_http_headers_as_record_headers = true
      forward_all_by_default = false
      include_headers = ["x-request-id", "x-tenant-id", "x-correlation-id"]

      name_mappings = {
        x-request-id = "trace-id"
        x-tenant-id = "tenant"
      }
      repeated_headers_behavior = "concatenate_by_comma"
    }
  }
  tags = []

  control_plane_id = konnect_gateway_control_plane.my_konnect_cp.id
  service = {
    id = konnect_gateway_service.my_service.id
  }
}

Copied!

This example requires the following variables to be added to your manifest. You can specify values at runtime by setting TF_VAR_name=value.

variable "kafka_topic" {
  type = string
}

Copied!

Add this section to your kong.yaml configuration file:

kong.yaml

Copied!

_format_version: "3.0"
plugins:
  - name: kafka-upstream
    route: routeName|Id
    config:
      topic: ${{ env "DECK_KAFKA_TOPIC" }}
      headers:
        forward_http_headers_as_record_headers: true
        forward_all_by_default: false
        include_headers:
        - x-request-id
        - x-tenant-id
        - x-correlation-id
        name_mappings:
          x-request-id: trace-id
          x-tenant-id: tenant
        repeated_headers_behavior: concatenate_by_comma

Make sure to replace the following placeholders with your own values:

routeName|Id: The id or name of the route the plugin configuration will target.

Make the following request:

curl -i -X POST http://localhost:8001/routes/{routeName|Id}/plugins/ \
    --header "Accept: application/json" \
    --header "Content-Type: application/json" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

routeName|Id: The id or name of the route the plugin configuration will target.

Make the following request:

curl -X POST https://{region}.api.konghq.com/v2/control-planes/{controlPlaneId}/core-entities/routes/{routeId}/plugins/ \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

region: Geographic region where your Kong Konnect is hosted and operates.
KONNECT_TOKEN: Your Personal Access Token (PAT) associated with your Konnect account.
controlPlaneId: The id of the control plane.
routeId: The id of the route the plugin configuration will target.

See the Konnect API reference to learn about region-specific URLs and personal access tokens.

echo "
apiVersion: configuration.konghq.com/v1
kind: KongPlugin
metadata:
  name: kafka-upstream
  namespace: kong
  annotations:
    kubernetes.io/ingress.class: kong
    konghq.com/tags: ''
config:
  topic: '$KAFKA_TOPIC'
  headers:
    forward_http_headers_as_record_headers: true
    forward_all_by_default: false
    include_headers:
    - x-request-id
    - x-tenant-id
    - x-correlation-id
    name_mappings:
      x-request-id: trace-id
      x-tenant-id: tenant
    repeated_headers_behavior: concatenate_by_comma
plugin: kafka-upstream
" | kubectl apply -f -

Copied!

Next, apply the KongPlugin resource by annotating the httproute or ingress resource:

kubectl annotate -n kong httproute  konghq.com/plugins=kafka-upstream

Copied!

kubectl annotate -n kong ingress  konghq.com/plugins=kafka-upstream

Copied!

Prerequisite: Configure your Personal Access Token

terraform {
  required_providers {
    konnect = {
      source  = "kong/konnect"
    }
  }
}

provider "konnect" {
  personal_access_token = "$KONNECT_TOKEN"
  server_url            = "https://us.api.konghq.com/"
}

Copied!

Add the following to your Terraform configuration to create a Konnect Gateway Plugin:

resource "konnect_gateway_plugin_kafka_upstream" "my_kafka_upstream" {
  enabled = true

  config = {
    topic = var.kafka_topic

    headers = {
      forward_http_headers_as_record_headers = true
      forward_all_by_default = false
      include_headers = ["x-request-id", "x-tenant-id", "x-correlation-id"]

      name_mappings = {
        x-request-id = "trace-id"
        x-tenant-id = "tenant"
      }
      repeated_headers_behavior = "concatenate_by_comma"
    }
  }
  tags = []

  control_plane_id = konnect_gateway_control_plane.my_konnect_cp.id
  route = {
    id = konnect_gateway_route.my_route.id
  }
}

Copied!

This example requires the following variables to be added to your manifest. You can specify values at runtime by setting TF_VAR_name=value.

variable "kafka_topic" {
  type = string
}

Copied!

Add this section to your kong.yaml configuration file:

kong.yaml

Copied!

_format_version: "3.0"
plugins:
  - name: kafka-upstream
    consumer: consumerName|Id
    config:
      topic: ${{ env "DECK_KAFKA_TOPIC" }}
      headers:
        forward_http_headers_as_record_headers: true
        forward_all_by_default: false
        include_headers:
        - x-request-id
        - x-tenant-id
        - x-correlation-id
        name_mappings:
          x-request-id: trace-id
          x-tenant-id: tenant
        repeated_headers_behavior: concatenate_by_comma

Make sure to replace the following placeholders with your own values:

consumerName|Id: The id or name of the consumer the plugin configuration will target.

Make the following request:

curl -i -X POST http://localhost:8001/consumers/{consumerName|Id}/plugins/ \
    --header "Accept: application/json" \
    --header "Content-Type: application/json" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

consumerName|Id: The id or name of the consumer the plugin configuration will target.

Make the following request:

curl -X POST https://{region}.api.konghq.com/v2/control-planes/{controlPlaneId}/core-entities/consumers/{consumerId}/plugins/ \
    --header "accept: application/json" \
    --header "Content-Type: application/json" \
    --header "Authorization: Bearer $KONNECT_TOKEN" \
    --data '
    {
      "name": "kafka-upstream",
      "config": {
        "topic": "'$KAFKA_TOPIC'",
        "headers": {
          "forward_http_headers_as_record_headers": true,
          "forward_all_by_default": false,
          "include_headers": [
            "x-request-id",
            "x-tenant-id",
            "x-correlation-id"
          ],
          "name_mappings": {
            "x-request-id": "trace-id",
            "x-tenant-id": "tenant"
          },
          "repeated_headers_behavior": "concatenate_by_comma"
        }
      },
      "tags": []
    }
    '

Copied!

Make sure to replace the following placeholders with your own values:

region: Geographic region where your Kong Konnect is hosted and operates.
KONNECT_TOKEN: Your Personal Access Token (PAT) associated with your Konnect account.
controlPlaneId: The id of the control plane.
consumerId: The id of the consumer the plugin configuration will target.

See the Konnect API reference to learn about region-specific URLs and personal access tokens.

echo "
apiVersion: configuration.konghq.com/v1
kind: KongPlugin
metadata:
  name: kafka-upstream
  namespace: kong
  annotations:
    kubernetes.io/ingress.class: kong
    konghq.com/tags: ''
config:
  topic: '$KAFKA_TOPIC'
  headers:
    forward_http_headers_as_record_headers: true
    forward_all_by_default: false
    include_headers:
    - x-request-id
    - x-tenant-id
    - x-correlation-id
    name_mappings:
      x-request-id: trace-id
      x-tenant-id: tenant
    repeated_headers_behavior: concatenate_by_comma
plugin: kafka-upstream
" | kubectl apply -f -

Copied!

Next, apply the KongPlugin resource by annotating the KongConsumer resource:

kubectl annotate -n kong kongconsumer CONSUMER_NAME konghq.com/plugins=kafka-upstream

Copied!

Prerequisite: Configure your Personal Access Token

terraform {
  required_providers {
    konnect = {
      source  = "kong/konnect"
    }
  }
}

provider "konnect" {
  personal_access_token = "$KONNECT_TOKEN"
  server_url            = "https://us.api.konghq.com/"
}

Copied!

Add the following to your Terraform configuration to create a Konnect Gateway Plugin:

resource "konnect_gateway_plugin_kafka_upstream" "my_kafka_upstream" {
  enabled = true

  config = {
    topic = var.kafka_topic

    headers = {
      forward_http_headers_as_record_headers = true
      forward_all_by_default = false
      include_headers = ["x-request-id", "x-tenant-id", "x-correlation-id"]

      name_mappings = {
        x-request-id = "trace-id"
        x-tenant-id = "tenant"
      }
      repeated_headers_behavior = "concatenate_by_comma"
    }
  }
  tags = []

  control_plane_id = konnect_gateway_control_plane.my_konnect_cp.id
  consumer = {
    id = konnect_gateway_consumer.my_consumer.id
  }
}

Copied!

This example requires the following variables to be added to your manifest. You can specify values at runtime by setting TF_VAR_name=value.

variable "kafka_topic" {
  type = string
}

Copied!

Kafka Upstream

Forward HTTP headers as Kafka record headers (allowlist mode)v3.15+

Prerequisites

Environment variables

Set up the plugin

Help us make these docs great!

Still need help?