Kong · Schema
KafkaUpstreamPluginConfig
API GatewayAI GatewayAI ConnectivityAgent GatewayEvent GatewayMCP RegistryService MeshLLMKafkaKonnectOpen Source
Properties
| Name | Type | Description |
|---|---|---|
| config | object | |
| consumer | object | If set, the plugin will activate only for requests where the specified has been authenticated. (Note that some plugins can not be restricted to consumers this way.). Leave unset for the plugin to acti |
| name | object | |
| protocols | array | A set of strings representing HTTP protocols. |
| route | object | If set, the plugin will only activate when receiving requests via the specified route. Leave unset for the plugin to activate regardless of the route being used. |
| service | object | If set, the plugin will only activate when receiving requests via one of the routes belonging to the specified Service. Leave unset for the plugin to activate regardless of the Service being matched. |
JSON Schema
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "#/components/schemas/KafkaUpstreamPluginConfig",
"title": "KafkaUpstreamPluginConfig",
"x-speakeasy-entity": "PluginKafkaUpstream",
"properties": {
"config": {
"type": "object",
"properties": {
"allowed_topics": {
"description": "The list of allowed topic names to which messages can be sent. The default topic configured in the `topic` field is always allowed, regardless of its inclusion in `allowed_topics`.",
"type": "array",
"items": {
"type": "string"
}
},
"authentication": {
"type": "object",
"properties": {
"mechanism": {
"description": "The SASL authentication mechanism. Supported options: `PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`.",
"type": "string",
"enum": [
"PLAIN",
"SCRAM-SHA-256",
"SCRAM-SHA-512"
]
},
"password": {
"description": "Password for SASL authentication.",
"type": "string",
"x-encrypted": true,
"x-referenceable": true
},
"strategy": {
"description": "The authentication strategy for the plugin, the only option for the value is `sasl`.",
"type": "string",
"enum": [
"sasl"
]
},
"tokenauth": {
"description": "Enable this to indicate `DelegationToken` authentication.",
"type": "boolean"
},
"user": {
"description": "Username for SASL authentication.",
"type": "string",
"x-encrypted": true,
"x-referenceable": true
}
}
},
"bootstrap_servers": {
"description": "Set of bootstrap brokers in a `{host: host, port: port}` list format.",
"type": "array",
"items": {
"properties": {
"host": {
"description": "A string representing a host name, such as example.com.",
"type": "string"
},
"port": {
"description": "An integer representing a port number between 0 and 65535, inclusive.",
"type": "integer",
"maximum": 65535,
"minimum": 0
}
},
"required": [
"host",
"port"
],
"type": "object"
}
},
"cluster_name": {
"description": "An identifier for the Kafka cluster. By default, this field generates a random string. You can also set your own custom cluster identifier. If more than one Kafka plugin is configured without a `cluster_name` (that is, if the default autogenerated value is removed), these plugins will use the same producer, and by extension, the same cluster. Logs will be sent to the leader of the cluster.",
"type": "string"
},
"forward_body": {
"description": "Include the request body in the message. At least one of these must be true: `forward_method`, `forward_uri`, `forward_headers`, `forward_body`.",
"type": "boolean",
"default": true
},
"forward_headers": {
"description": "Include the request headers in the message. At least one of these must be true: `forward_method`, `forward_uri`, `forward_headers`, `forward_body`.",
"type": "boolean",
"default": false
},
"forward_method": {
"description": "Include the request method in the message. At least one of these must be true: `forward_method`, `forward_uri`, `forward_headers`, `forward_body`.",
"type": "boolean",
"default": false
},
"forward_uri": {
"description": "Include the request URI and URI arguments (as in, query arguments) in the message. At least one of these must be true: `forward_method`, `forward_uri`, `forward_headers`, `forward_body`.",
"type": "boolean",
"default": false
},
"keepalive": {
"description": "Keepalive timeout in milliseconds.",
"type": "integer",
"default": 60000
},
"keepalive_enabled": {
"type": "boolean",
"default": false
},
"key_query_arg": {
"description": "The request query parameter name that contains the Kafka message key. If specified, messages with the same key will be sent to the same Kafka partition, ensuring consistent ordering.",
"type": "string"
},
"message_by_lua_functions": {
"description": "The Lua functions that manipulates the message being sent to the Kafka topic.",
"type": "array",
"items": {
"type": "string"
}
},
"producer_async": {
"description": "Flag to enable asynchronous mode.",
"type": "boolean",
"default": true
},
"producer_async_buffering_limits_messages_in_memory": {
"description": "Maximum number of messages that can be buffered in memory in asynchronous mode.",
"type": "integer",
"default": 50000
},
"producer_async_flush_timeout": {
"description": "Maximum time interval in milliseconds between buffer flushes in asynchronous mode.",
"type": "integer",
"default": 1000
},
"producer_request_acks": {
"description": "The number of acknowledgments the producer requires the leader to have received before considering a request complete. Allowed values: 0 for no acknowledgments; 1 for only the leader; and -1 for the full ISR (In-Sync Replica set).",
"type": "integer",
"default": 1,
"enum": [
-1,
0,
1
]
},
"producer_request_limits_bytes_per_request": {
"description": "Maximum size of a Produce request in bytes.",
"type": "integer",
"default": 1048576
},
"producer_request_limits_messages_per_request": {
"description": "Maximum number of messages to include into a single producer request.",
"type": "integer",
"default": 200
},
"producer_request_retries_backoff_timeout": {
"description": "Backoff interval between retry attempts in milliseconds.",
"type": "integer",
"default": 100
},
"producer_request_retries_max_attempts": {
"description": "Maximum number of retry attempts per single Produce request.",
"type": "integer",
"default": 10
},
"producer_request_timeout": {
"description": "Time to wait for a Produce response in milliseconds.",
"type": "integer",
"default": 2000
},
"schema_registry": {
"description": "The plugin-global schema registry configuration. This can be overwritten by the topic configuration.",
"type": "object",
"properties": {
"confluent": {
"type": "object",
"properties": {
"authentication": {
"type": "object",
"properties": {
"basic": {
"type": "object",
"properties": {
"password": {
"type": "string",
"x-encrypted": true,
"x-referenceable": true
},
"username": {
"type": "string",
"x-encrypted": true,
"x-referenceable": true
}
},
"required": [
"password",
"username"
]
},
"mode": {
"description": "Authentication mode to use with the schema registry.",
"type": "string",
"default": "none",
"enum": [
"basic",
"none",
"oauth2"
]
},
"oauth2": {
"type": "object",
"properties": {
"audience": {
"description": "List of audiences passed to the IdP when obtaining a new token.",
"type": "array",
"items": {
"type": "string"
},
"default": []
},
"client_id": {
"description": "The client ID for the application registration in the IdP.",
"type": "string",
"x-encrypted": true,
"x-referenceable": true
},
"client_secret": {
"description": "The client secret for the application registration in the IdP.",
"type": "string",
"x-encrypted": true,
"x-referenceable": true
},
"grant_type": {
"description": "The OAuth grant type to be used.",
"type": "string",
"default": "client_credentials",
"enum": [
"client_credentials",
"password"
]
},
"password": {
"description": "The password to use if `config.oauth.grant_type` is set to `password`.",
"type": "string",
"x-encrypted": true,
"x-referenceable": true
},
"scopes": {
"description": "List of scopes to request from the IdP when obtaining a new token.",
"type": "array",
"items": {
"type": "string"
},
"default": [
"openid"
]
},
"token_endpoint": {
"description": "The token endpoint URI.",
"type": "string"
},
"token_headers": {
"description": "Extra headers to be passed in the token endpoint request.",
"type": "object",
"additionalProperties": {
"type": "string",
"x-referenceable": true
}
},
"token_post_args": {
"description": "Extra post arguments to be passed in the token endpoint request.",
"type": "object",
"additionalProperties": {
"type": "string",
"x-referenceable": true
}
},
"username": {
"description": "The username to use if `config.oauth.grant_type` is set to `password`.",
"type": "string",
"x-encrypted": true,
"x-referenceable": true
}
},
"required": [
"token_endpoint"
]
},
"oauth2_client": {
"type": "object",
"properties": {
"auth_method": {
"description": "The authentication method used in client requests to the IdP. Supported values are: `client_secret_basic` to send `client_id` and `client_secret` in the `Authorization: Basic` header, `client_secret_post` to send `client_id` and `client_secret` as part of the request body, or `client_secret_jwt` to send a JWT signed with the `client_secret` using the client assertion as part of the body.",
"type": "string",
"default": "client_secret_post",
"enum": [
"client_secret_basic",
"client_secret_jwt",
"client_secret_post",
"none"
]
},
"client_secret_jwt_alg": {
"description": "The algorithm to use with JWT when using `client_secret_jwt` authentication.",
"type": "string",
"default": "HS512",
"enum": [
"HS256",
"HS512"
]
},
"http_proxy": {
"description": "The proxy to use when making HTTP requests to the IdP.",
"type": "string"
},
"http_proxy_authorization": {
"description": "The `Proxy-Authorization` header value to be used with `http_proxy`.",
"type": "string"
},
"http_version": {
"description": "The HTTP version used for requests made by this plugin. Supported values: `1.1` for HTTP 1.1 and `1.0` for HTTP 1.0.",
"type": "number",
"default": 1.1
},
"https_proxy": {
"description": "The proxy to use when making HTTPS requests to the IdP.",
"type": "string"
},
"https_proxy_authorization": {
"description": "The `Proxy-Authorization` header value to be used with `https_proxy`.",
"type": "string"
},
"keep_alive": {
"description": "Whether to use keepalive connections to the IdP.",
"type": "boolean",
"default": true
},
"no_proxy": {
"description": "A comma-separated list of hosts that should not be proxied.",
"type": "string"
},
"ssl_verify": {
"description": "Whether to verify the certificate presented by the IdP when using HTTPS.",
"type": "boolean",
"default": true
},
"timeout": {
"description": "Network I/O timeout for requests to the IdP in milliseconds.",
"type": "integer",
"default": 10000,
"maximum": 2147483646,
"minimum": 0
}
}
}
}
},
"key_schema": {
"type": "object",
"properties": {
"schema_version": {
"description": "The schema version to use for serialization/deserialization. Use 'latest' to always fetch the most recent version.",
"type": "string"
},
"subject_name": {
"description": "The name of the subject",
"type": "string"
}
}
},
"ssl_verify": {
"description": "Set to false to disable SSL certificate verification when connecting to the schema registry.",
"type": "boolean",
"default": true
},
"ttl": {
"description": "The TTL in seconds for the schema registry cache.",
"type": "number",
"maximum": 3600,
"minimum": 0
},
"url": {
"description": "The URL of the schema registry.",
"type": "string"
},
"value_schema": {
"type": "object",
"properties": {
"schema_version": {
"description": "The schema version to use for serialization/deserialization. Use 'latest' to always fetch the most recent version.",
"type": "string"
},
"subject_name": {
"description": "The name of the subject",
"type": "string"
}
}
}
}
}
}
},
"security": {
"type": "object",
"properties": {
"certificate_id": {
"description": "UUID of certificate entity for mTLS authentication.",
"type": "string"
},
"ssl": {
"description": "Enables TLS.",
"type": "boolean"
},
"ssl_verify": {
"description": "When using TLS, this option enables verification of the certificate presented by the server.",
"type": "boolean",
"default": true
}
}
},
"timeout": {
"description": "Socket timeout in milliseconds.",
"type": "integer",
"default": 10000
},
"topic": {
"description": "The default Kafka topic to publish to if the query parameter defined in the `topics_query_arg` does not exist in the request",
"type": "string"
},
"topics_query_arg": {
"description": "The request query parameter name that contains the topics to publish to",
"type": "string"
}
},
"required": [
"topic"
]
},
"consumer": {
"description": "If set, the plugin will activate only for requests where the specified has been authenticated. (Note that some plugins can not be restricted to consumers this way.). Leave unset for the plugin to activate regardless of the authenticated Consumer.",
"type": "object",
"additionalProperties": false,
"properties": {
"id": {
"type": "string"
}
}
},
"name": {
"const": "kafka-upstream"
},
"protocols": {
"description": "A set of strings representing HTTP protocols.",
"type": "array",
"items": {
"enum": [
"grpc",
"grpcs",
"http",
"https"
],
"type": "string"
},
"format": "set",
"default": [
"grpc",
"grpcs",
"http",
"https"
]
},
"route": {
"description": "If set, the plugin will only activate when receiving requests via the specified route. Leave unset for the plugin to activate regardless of the route being used.",
"type": "object",
"additionalProperties": false,
"properties": {
"id": {
"type": "string"
}
}
},
"service": {
"description": "If set, the plugin will only activate when receiving requests via one of the routes belonging to the specified Service. Leave unset for the plugin to activate regardless of the Service being matched.",
"type": "object",
"additionalProperties": false,
"properties": {
"id": {
"type": "string"
}
}
}
},
"required": [
"config"
]
}