Consumes messages from an Apache Kafka topic. Every message runs through the flow of this API and is sent to its target, like a request that came in over HTTP. The target can be an HTTP backend or another API using internal://.

A message is delivered when the target answers with a status below 400. A client error skips the message, since sending it again would not help. A server error, an unreachable target, 408 Request Timeout and 429 Too Many Requests deliver the message again until it succeeds or maxRetries is reached; until then, the following messages of the same partition wait.

By default a message is acknowledged to Kafka only after it was delivered or skipped, so no message is lost, but it can be delivered more than once and the target should handle duplicates. With automatic acknowledgement, Kafka commits received messages in the background instead.

The API does not support hot deployment and retryInit. Hot deployment is on by default, so turn it off with hotDeploy: false under configuration; retryInit is off by default.

The API needs the Membrane Kafka Extension. Without the extension Membrane does not start. Contact predic8 for the extension.

Example Configuration

kafkaConsumer:
   name: orders-in
   bootstrapServers: kafka.example.com:9092
   topic: orders
   groupId: membrane
   flow:
     - setHeader:
         name: X-Source
         value: kafka
   target:
     url: https://orders.example.com/orders

kafkaConsumer:
   name: orders-in
   bootstrapServers: kafka.example.com:9092
   topic: orders
   groupId: membrane
   flow:
     - setHeader:
         name: X-Source
         value: kafka
   target:
     url: https://orders.example.com/orders

Syntax

kafkaConsumer:
  bootstrapServers: <string> # required
  groupId: <string> # required
  topic: <string> # required
  '$ref': <string>
  ackMode: auto
  autoOffsetReset: <string>
  clientId: <string>
  fetchMaxWaitMs: <int>
  fetchMinBytes: <int>
  flow:
    - <flow>
  host: <string>
  ip: <string>
  maxPollIntervalMs: <int>
  maxPollRecords: <int>
  maxRetries: <int>
  name: <string>
  path: {}
  port: <int>
  saslJaasConfig: <string>
  saslMechanism: <string>
  securityProtocol: <string>
  sessionTimeoutMs: <int>
  ssl: {}
  target: {}

kafkaConsumer:
  bootstrapServers: <string> # required
  groupId: <string> # required
  topic: <string> # required
  '$ref': <string>
  ackMode: auto
  autoOffsetReset: <string>
  clientId: <string>
  fetchMaxWaitMs: <int>
  fetchMinBytes: <int>
  flow:
    - <flow>
  host: <string>
  ip: <string>
  maxPollIntervalMs: <int>
  maxPollRecords: <int>
  maxRetries: <int>
  name: <string>
  path: {}
  port: <int>
  saslJaasConfig: <string>
  saslMechanism: <string>
  securityProtocol: <string>
  sessionTimeoutMs: <int>
  ssl: {}
  target: {}

Attributes

Name	Required	Default	Description	Examples
$ref	false	-	Reference a component defined under components.	-
ackMode	false	client	When a message is acknowledged to Kafka. With `client`, a message is acknowledged after it was delivered or skipped. With `auto`, Kafka acknowledges received messages periodically in the background, so a message counts as consumed once it is received, and it can be lost if Membrane stops before it is delivered.	auto
autoOffsetReset	false	latest	Where to start reading when the group has no committed offset.	earliest \| latest
bootstrapServers	true	-	Comma separated list of Kafka brokers used for the initial connection.	kafka1.example.com:9092,kafka2.example.com:9092
clientId	false	-	Client id sent to the brokers, useful for logging and quotas on the Kafka side.	membrane
fetchMaxWaitMs	false	500	Maximum time in milliseconds the broker waits for fetchMinBytes to be available.	-
fetchMinBytes	false	1	Minimum amount of data in bytes the broker returns for a fetch.	-
groupId	true	-	Consumer group the API joins. Consumers of the same group share the partitions of the topic.	membrane
host	false	not set	Restricts to requests whose `Host` header matches one of the given hostnames. Separate multiple hostnames with spaces. The asterisk `*` matches any number of characters, including zero, for basic globbing.	predic8.de *.predic8.de
ip	false	not set	If present, binds the port only on the specified IP. Useful for hosts with multiple IP addresses.	127.0.0.1
maxPollIntervalMs	false	300000	Maximum time in milliseconds between two polls, i.e. the time available to process a batch of messages.	-
maxPollRecords	false	500	Maximum number of records returned by a single poll.	-
maxRetries	false	-1	How often a message is delivered again after a server error, an unreachable target, 408 or 429. When the last retry fails too, an error is logged and the message is skipped. A negative value retries until the message is delivered.	10
name	false	By default, a name will be automatically generated from the target, openapi, etc.	The name of the proxy or API.	-
port	false	80	The port Membrane listens on for incoming connections.	8080
saslJaasConfig	false	-	JAAS login configuration for SASL authentication.	org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="secret";
saslMechanism	false	-	SASL mechanism for authentication, used with the security protocols SASL_PLAINTEXT and SASL_SSL.	PLAIN \| SCRAM-SHA-256 \| SCRAM-SHA-512
securityProtocol	false	-	Protocol used to talk to the brokers. Kafka default is PLAINTEXT.	PLAINTEXT \| SSL \| SASL_PLAINTEXT \| SASL_SSL
sessionTimeoutMs	false	45000	Time in milliseconds after which a consumer without heartbeat is considered dead.	-
topic	true	-	Topic the messages are read from.	orders

Child Structure

Element	Cardinality	Description
flow	0..*
path	0..1	Restricts to requests whose path matches, either by prefix (starts-with) or, when enabled, by regular expression. In a `soapProxy`, setting a path also rewrites the paths of SOAP requests and the service addresses in the WSDL accordingly.
ssl	0..1	Configures a TLS context: the identity (key and certificate) presented to the peer, the certificates trusted from it, and handshake parameters such as protocol, ciphers, and client-certificate policy. Attach it to a `serviceProxy`/`api` to terminate inbound TLS, or to a `target` to make an outbound connection over TLS; most attributes and child elements apply to both directions, but a few - such as `clientAuth` and `useAsDefault` - only take effect on an inbound context. See `tutorials/ssl-tls/10-TLS-Termination.yaml` for inbound termination and `tutorials/ssl-tls/20-Central-SSL-Config.yaml` for sharing one `ssl` across several APIs via `$ref`.
target	0..1	The backend server that requests are forwarded to after passing through the flow.

Can be used in

No parent elements.