Skip to content

Latest commit

 

History

History
451 lines (423 loc) · 63.9 KB

README_kafka.md

File metadata and controls

451 lines (423 loc) · 63.9 KB

gateleen-kafka

This module provides a handler to forward http requests to Apache Kafka using the native Vert.x Kafka client.

For now, only sending messages to Kafka (via KafkaProducer) is available. Consuming messages from Kafka (via KafkaConsumer) is not implemented.

Configuration

To configure this module, the following paths have to be configured during setup of the KafkaHandler class:

Path Description
configResourceUri The path to the topics configuration resource. Example: /playground/server/admin/v1/kafka/topicsConfig
streamingPath The path to handle kafka message requests. Example: /playground/server/streaming/

Topic configuration resource

This resource holds the KafkaProducer configuration values for topics. The path to this resource is the configResourceUri mentioned above.

Modifications (PUT requests) to this resource are validated against the schema gateleen_kafka_topic_configuration_schema.

The following example shows a topic configuration resource with configurations for two topics. The order of the topics is important. Topics are tried to match against the topic names (regex) from top to bottom. The first topic name matching the provided topic is used.

{
	"my.topic.*": {
		"bootstrap.servers": "localhost:9092",
		"key.serializer": "org.apache.kafka.common.serialization.StringSerializer",
		"value.serializer": "org.apache.kafka.common.serialization.StringSerializer"
	},
	".+": {
		"bootstrap.servers": "localhost:9093",
		"key.serializer": "org.apache.kafka.common.serialization.StringSerializer",
		"value.serializer": "org.apache.kafka.common.serialization.StringSerializer"
	}
}

In the example above, a topic called my.topic.abc would use the first configuration entry. A topic called some.other.topic would use the second configuration entry.

Topic configuration values

The following topic configuration values are required:

Configuration value Description
bootstrap.servers A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. This list should be in the form host1:port1,host2:port2,.... Since these servers are just used for the initial connection to discover the full cluster membership (which may change dynamically), this list need not contain the full set of servers (you may want more than one, though, in case a server is down).
key.serializer Serializer class for key that implements the org.apache.kafka.common.serialization.Serializer interface.
value.serializer Serializer class for value that implements the org.apache.kafka.common.serialization.Serializer interface.

Besides these required configuration values, additional string values can be added. See documentation from Apache Kafka here.

Usage

To use the gateleen-kafka module, the KafkaHandler class has to be initialized as described in the configuration section. Also the KafkaHandler has to be integrated in the "MainVerticle" handling all incoming requests. See Playground Server and Runconfig.

The following sequence diagram shows the setup of the "MainVerticle". The streamingPath (KafkaHandler) is configured to /playground/server/streaming/

       ┌─┐                                                                                                                                      
       ║"│                                                                                                                                      
       └┬┘                                                                                                                                      
       ┌┼┐                                                                                                                                      
        │            ┌────────────┐                                                    ┌────────────┐          ┌─────────────┐          ┌──────┐
       ┌┴┐           │MainVerticle│                                                    │KafkaHandler│          │OtherHandlers│          │Router│
      User           └─────┬──────┘                                                    └─────┬──────┘          └──────┬──────┘          └──┬───┘
       │     Request      ┌┴┐                                                                │                        │                    │    
       │ ────────────────>│ │                                                                │                        │                    │    
       │                  │ │                                                                │                        │                    │    
       │                  │ │[uri == "/playground/server/streaming/myTopic"] handle(request)┌┴┐                       │                    │    
       │                  │ │ ─────────────────────────────────────────────────────────────>│ │                       │                    │    
       │                  │ │                                                               └┬┘                       │                    │    
       │                  │ │                              true                              │                        │                    │    
       │                  │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─                        │                    │    
       │                  │ │                                                                │                        │                    │    
       │                  │ │                       [request not yet handled] handle(request)│                       ┌┴┐                   │    
       │                  │ │ ──────────────────────────────────────────────────────────────────────────────────────>│ │                   │    
       │                  │ │                                                                │                       └┬┘                   │    
       │                  │ │                                  handled (true/false)          │                        │                    │    
       │                  │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─│                    │    
       │                  │ │                                                                │                        │                    │    
       │                  │ │                                 [no handler handled request] route(request)             │                   ┌┴┐  
       │                  │ │ ───────────────────────────────────────────────────────────────────────────────────────────────────────────>│ │  
       │                  │ │                                                                │                        │                   └┬┘  
       │                  │ │                                                                │                        │                    │    
       │                  │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─    
       │                  └┬┘                                                                │                        │                    │    
       │      Response     │                                                                 │                        │                    │    
       │ <─ ─ ─ ─ ─ ─ ─ ─ ─                                                                  │                        │                    │    
       │                   │                                                                 │                        │                    │    
       │                   │                                                                 │                        │                    │    

Sending messages to Kafka

To send messages to Kafka through the KafkaHandler, a POST request to

/playground/server/streaming/myTopic

is used. The last url segment myTopic defines the topic to send the messages to.

The Payload of the POST request must have the following structure:

{
  "records": [
    {
      "key": "key_1",
      "value": {},
      "headers": {}
    }
  ]
}
Property Required Description
records yes An array holding 1 ore more record objects
value yes The actual message
key no Messages with the same key will be sent to the same partition. When no key is defined the destination partition is defined by a round robin algorithm
headers no Key/Value pairs (String) to add as headers to the message

Example:

{
  "records": [
    {
      "key": "key_1",
      "value": {
        "metadata": {
          "techId": "071X15004960860907613"
        },
        "event": {
          "actionTime": "2019-06-18T14:28:27.617+02:00",
          "sending": {
            "identCode": "777",
            "capturingType": "1"
          }
        }
      },
      "headers": {
        "x-header-a": "value-a",
        "x-header-b": "value-b"
      }
    }
  ]
}

Transactional transmission

When sending multiple messages (having more than 1 entry in the records array), transactional transmission is not guaranteed since, Vert.x Kafka client does not yet provide transaction support.

See the following issues requesting this feature:

Delegates vs Kafka

When working with Kafka, the gateleen-delegate module is most likely be used too, because delegates can transform an incoming http request body to the structure needed by Kafka.

The following delegate definition transforms the incoming payload to the correct structure and forwards it by making the following request, which will be handled by the KafkaHandler:

POST /playground/server/streaming/myTopic

Delegate definition:

{
  "methods": [
    "PUT"
  ],
  "pattern": ".*/execution/([^/]+)",
  "requests": [
    {
      "method": "POST",
      "uri": "/playground/server/streaming/myTopic",
      "transformWithMetadata": [
        {
          "operation": "shift",
          "spec": {
            "urlParts": {
              "1": "records[0].value.metadata.techId"
            },
            "headers": {
              "x-abc": "records[0].value.metadata.x-abc"
            },
            "payload": {
              "@": "records[0].value.data",
              "node": {
                "id": [
                  "records[0].key",
                  "records[0].value.&"
                ]
              }
            }
          }
        }
      ]
    }
  ]
}

With this delegate definition, the incoming http request

{
  "urlParts": [
    "/some/test/url/xyz",
    "xyz"
  ],
  "headers": {
    "x-abc": "x",
    "x-def": "x,y",
    "x-ghi": "z"
  },
  "payload": {
    "parent": "10",
    "node": {
      "id": "Key_123456",
      "type": "x",
      "isLeaf": false
    }
  }
}

will be transformed into

{
  "records": [
    {
      "value": {
        "metadata": {
          "techId": "xyz",
          "x-abc": "x"
        },
        "data": {
          "parent": "10",
          "node": {
            "id": "Key_123456",
            "type": "x",
            "isLeaf": false
          }
        },
        "id": "Key_123456"
      },
      "key": "Key_123456"
    }
  ]
}

which is the required structure for the KafkaHandler to extract the message(s) from.

Main processes sequence diagrams

The following sections contain sequence diagrams showing the inner workings of the most critical processes.

Initialization KafkaHandler

This sequence diagrams shows the initialization of the KafkaHandler on server startup:

                      ┌─┐                                                                                                                                                                                                 
                      ║"│                                                                                                                                                                                                 
                      └┬┘                                                                                                                                                                                                 
                      ┌┼┐                                                                                                                                                                                                 
                       │             ┌────────────┐                             ┌────────────────────────────┐          ┌───────────────┐          ┌────────────────────────┐          ┌───────────────────────┐          
                      ┌┴┐            │KafkaHandler│                             │ConfigurationResourceManager│          │ResourceStorage│          │KafkaConfigurationParser│          │KafkaProducerRepository│          
                     Setup           └─────┬──────┘                             └─────────────┬──────────────┘          └───────┬───────┘          └───────────┬────────────┘          └───────────┬───────────┘          
                       │  initialize()    ┌┴┐                                                 │                                 │                              │                                   │                      
                       │─────────────────>│ │                                                 │                                 │                              │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  │ │   registerResource(configResourceUri, schema)   │                                 │                              │                                   │                      
                       │                  │ │ ───────────────────────────────────────────────>│                                 │                              │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  │ │    registerObserver(this, configResourceUri)    │                                 │                              │                                   │                      
                       │                  │ │ ───────────────────────────────────────────────>│                                 │                              │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  │ │    getRegisteredResource(configResourceUri)    ┌┴┐                                │                              │                                   │                      
                       │                  │ │ ──────────────────────────────────────────────>│ │                                │                              │                                   │                      
                       │                  │ │                                                │ │                                │                              │                                   │                      
                       │                  │ │                                                │ │    get(configResourceUri)     ┌┴┐                             │                                   │                      
                       │                  │ │                                                │ │ ────────────────────────────> │ │                             │                                   │                      
                       │                  │ │                                                │ │                               └┬┘                             │                                   │                      
                       │                  │ │                                                │ │        Optional<Buffer>        │                              │                                   │                      
                       │                  │ │                                                │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                              │                                   │                      
                       │                  │ │                                                └┬┘                                │                              │                                   │                      
                       │                  │ │                Optional<Buffer>                 │                                 │                              │                                   │                      
                       │                  │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─│                                 │                              │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
          ╔══════╤═════╪══════════════════╪═╪═════════════════════════════════════════════════╪══╗                              │                              │                                   │                      
          ║ ALT  │  missing configuration case                                                │  ║                              │                              │                                   │                      
          ╟──────┘     │                  │ │                                                 │  ║                              │                              │                                   │                      
          ║            │                  │ │────┐                                            │  ║                              │                              │                                   │                      
          ║            │                  │ │    │ [configuration resource missing] log error │  ║                              │                              │                                   │                      
          ║            │                  │ │<───┘                                            │  ║                              │                              │                                   │                      
          ║            │                  └┬┘                                                 │  ║                              │                              │                                   │                      
          ║            │                   │                                                  │  ║                              │                              │                                   │                      
          ║            │<─ ─ ─ ─ ─ ─ ─ ─ ─ │                                                  │  ║                              │                              │                                   │                      
          ╚════════════╪═══════════════════╪══════════════════════════════════════════════════╪══╝                              │                              │                                   │                      
                       │                   │                                                  │                                 │                              │                                   │                      
                       │                  ┌┴┐                                          parse(configurationResource)             │                             ┌┴┐                                  │                      
                       │                  │ │ ───────────────────────────────────────────────────────────────────────────────────────────────────────────────>│ │                                  │                      
                       │                  │ │                                                 │                                 │                             └┬┘                                  │                      
                       │                  │ │                                             List<KafkaConfiguration>              │                              │                                   │                      
                       │                  │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  │ │                                                 │                    closeAll()   │                              │                                   │                      
                       │                  │ │ ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────>│                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │   ╔═══════╤══════╪═╪═════════════════════════════════════════════════╪═════════════════════════════════╪══════════════════════════════╪═══════════════════════════════════╪═════════════════════╗
                       │   ║ LOOP  │  for every KafkaConfiguration                            │                                 │                              │                                   │                     ║
                       │   ╟───────┘      │ │                                                 │                                 │                              │                                   │                     ║
                       │   ║              │ │                                                 │      addKafkaProducer(kafkaConfiguration)                      │                                  ┌┴┐                    ║
                       │   ║              │ │ ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────> │ │                    ║
                       │   ║              │ │                                                 │                                 │                              │                                  │ │                    ║
                       │   ║              │ │                                                 │                                 │                           ╔══╧════════════════════════════════╗ │ │                    ║
                       │   ║              │ │                                                 │                                 │                           ║instantiate Producer and store it ░║ │ │                    ║
                       │   ║              │ │                                                 │                                 │                           ║in a map with key 'topic-pattern'  ║ │ │                    ║
                       │   ║              │ │                                                 │                                 │                           ║and value producer                 ║ │ │                    ║
                       │   ║              │ │                                                 │                                 │                           ╚══╤════════════════════════════════╝ └┬┘                    ║
                       │   ║              │ │                                                 │                                 │                              │                                   │                     ║
                       │   ║              │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                     ║
                       │   ╚══════════════╪═╪═════════════════════════════════════════════════╪═════════════════════════════════╪══════════════════════════════╪═══════════════════════════════════╪═════════════════════╝
                       │                  │ │                                                 │                                 │                              │                                   │                      
                       │                  └┬┘                                                 │                                 │                              │                                   │                      

Update topic configuration resource

This sequence diagrams shows the process when updating the topic configuration resource:

     ┌────────────────────────────┐                                ┌────────────┐          ┌────────────────────────┐          ┌───────────────────────┐          
     │ConfigurationResourceManager│                                │KafkaHandler│          │KafkaConfigurationParser│          │KafkaProducerRepository│          
     └─────────────┬──────────────┘                                └─────┬──────┘          └───────────┬────────────┘          └───────────┬───────────┘          
                   │resourceChanged(String resourceUri, String resource)┌┴┐                            │                                   │                      
                   │ ──────────────────────────────────────────────────>│ │                            │                                   │                      
                   │                                                    │ │                            │                                   │                      
                   │                                                    │ │      parse(resource)      ┌┴┐                                  │                      
                   │                                                    │ │ ─────────────────────────>│ │                                  │                      
                   │                                                    │ │                           └┬┘                                  │                      
                   │                                                    │ │  List<KafkaConfiguration>  │                                   │                      
                   │                                                    │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                                   │                      
                   │                                                    │ │                            │                                   │                      
                   │                                                    │ │                           closeAll()                           │                      
                   │                                                    │ │ ──────────────────────────────────────────────────────────────>│                      
                   │                                                    │ │                            │                                   │                      
                   │                                                    │ │                            │                                   │                      
                   │                                     ╔═══════╤══════╪═╪════════════════════════════╪═══════════════════════════════════╪═════════════════════╗
                   │                                     ║ LOOP  │  for every KafkaConfiguration       │                                   │                     ║
                   │                                     ╟───────┘      │ │                            │                                   │                     ║
                   │                                     ║              │ │             addKafkaProducer(kafkaConfiguration)              ┌┴┐                    ║
                   │                                     ║              │ │ ────────────────────────────────────────────────────────────> │ │                    ║
                   │                                     ║              │ │                            │                                  │ │                    ║
                   │                                     ║              │ │                         ╔══╧════════════════════════════════╗ │ │                    ║
                   │                                     ║              │ │                         ║instantiate Producer and store it ░║ │ │                    ║
                   │                                     ║              │ │                         ║in a map with key 'topic-pattern'  ║ │ │                    ║
                   │                                     ║              │ │                         ║and value producer                 ║ │ │                    ║
                   │                                     ║              │ │                         ╚══╤════════════════════════════════╝ └┬┘                    ║
                   │                                     ║              │ │                            │                                   │                     ║
                   │                                     ║              │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                     ║
                   │                                     ╚══════════════╪═╪════════════════════════════╪═══════════════════════════════════╪═════════════════════╝
                   │                                                    │ │                            │                                   │                      
                   │                                                    └┬┘                            │                                   │                      

Sending messages to Kafka

This sequence diagrams shows the process when messages are sent to Kafka:

                    ┌────────────┐          ┌────────────┐           ┌───────────────────┐          ┌───────────────────────┐          ┌──────────────────────────┐               ┌──────────────────┐                         
                    │MainVerticle│          │KafkaHandler│           │KafkaTopicExtractor│          │KafkaProducerRepository│          │KafkaProducerRecordBuilder│               │KafkaMessageSender│                         
                    └─────┬──────┘          └─────┬──────┘           └─────────┬─────────┘          └───────────┬───────────┘          └────────────┬─────────────┘               └────────┬─────────┘                         
                          │   handle(request)    ┌┴┐                           │                                │                                   │                                      │                                   
                          │ ────────────────────>│ │                           │                                │                                   │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │                      │ │extractTopic(request.uri) ┌┴┐                               │                                   │                                      │                                   
                          │                      │ │ ───────────────────────> │ │                               │                                   │                                      │                                   
                          │                      │ │                          └┬┘                               │                                   │                                      │                                   
                          │                      │ │     topic (optional)      │                                │                                   │                                      │                                   
                          │                      │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─│                                │                                   │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
          ╔══════╤════════╪══════════════════════╪═╪═══════════════╗           │                                │                                   │                                      │                                   
          ║ ALT  │  topic is missing             │ │               ║           │                                │                                   │                                      │                                   
          ╟──────┘        │                      └┬┘               ║           │                                │                                   │                                      │                                   
          ║               │    400 Bad Request    │                ║           │                                │                                   │                                      │                                   
          ║               │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─                 ║           │                                │                                   │                                      │                                   
          ╚═══════════════╪═══════════════════════╪════════════════╝           │                                │                                   │                                      │                                   
                          │                       │                            │                                │                                   │                                      │                                   
                          │                      ┌┴┐             findMatchingKafkaProducer(topic)              ┌┴┐                                  │                                      │                                   
                          │                      │ │ ────────────────────────────────────────────────────────> │ │                                  │                                      │                                   
                          │                      │ │                           │                               └┬┘                                  │                                      │                                   
                          │                      │ │                    producer (optional)                     │                                   │                                      │                                   
                          │                      │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                                   │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
          ╔══════╤════════╪══════════════════════╪═╪═══════════════╗           │                                │                                   │                                      │                                   
          ║ ALT  │  no matching producer         │ │               ║           │                                │                                   │                                      │                                   
          ╟──────┘        │                      └┬┘               ║           │                                │                                   │                                      │                                   
          ║               │     404 Not Found     │                ║           │                                │                                   │                                      │                                   
          ║               │ <──────────────────────                ║           │                                │                                   │                                      │                                   
          ╚═══════════════╪═══════════════════════╪════════════════╝           │                                │                                   │                                      │                                   
                          │                       │                            │                                │                                   │                                      │                                   
                          │                      ┌┴┐                           │ buildRecords(topic, request.payload)                              ┌┴┐                                     │                                   
                          │                      │ │ ─────────────────────────────────────────────────────────────────────────────────────────────>│ │                                     │                                   
                          │                      │ │                           │                                │                                  │ │                                     │                                   
                          │                      │ │                           │                                │                                  │ │                                     │                                   
                          │                      │ │                           │                                │  ╔═══════╤═══════════════════════╪═╪═════════════════════════════════════╪══════════════╗                    
                          │                      │ │                           │                                │  ║ LOOP  │  for entry in records array                                   │              ║                    
                          │                      │ │                           │                                │  ╟───────┘                       │ │                                     │              ║                    
                          │                      │ │                           │                                │  ║                               │ │────┐                                │              ║                    
                          │                      │ │                           │                                │  ║                               │ │    │ extract 'key' from payload     │              ║                    
                          │                      │ │                           │                                │  ║                               │ │<───┘                                │              ║                    
                          │                      │ │                           │                                │  ║                               │ │                                     │              ║                    
                          │                      │ │                           │                                │  ║                               │ │  ╔══════════════════════════════════╧════════╗     ║                    
                          │                      │ │                           │                                │  ║                               │ │  ║See vertx-kafka-client documentation      ░║     ║                    
                          │                      │ │                           │                                │  ║                               │ │  ║on how to set key to record. If no key in  ║     ║                    
                          │                      │ │                           │                                │  ║                               │ │  ║payload found, use default (round robin)   ║     ║                    
                          │                      │ │                           │                                │  ║                               │ │  ╚══════════════════════════════════╤════════╝     ║                    
                          │                      │ │                           │                                │  ║                               │ │────┐                                │              ║                    
                          │                      │ │                           │                                │  ║                               │ │    │ extract 'value' from payload   │              ║                    
                          │                      │ │                           │                                │  ║                               │ │<───┘                                │              ║                    
                          │                      │ │                           │                                │  ║                               │ │                                     │              ║                    
                          │                      │ │                           │                                │  ║                               │ │────┐                                │              ║                    
                          │                      │ │                           │                                │  ║                               │ │    │ extract 'headers' from payload │              ║                    
                          │                      │ │                           │                                │  ║                               │ │<───┘                                │              ║                    
                          │                      │ │                           │                                │  ║                               │ │                                     │              ║                    
                          │                      │ │                           │                                │  ║                               │ │                                     │              ║                    
                          │                      │ │                           │                                │  ║         ╔═══════╤═════════════╪═╪═════════════════════════════════════╪═╗            ║                    
                          │                      │ │                           │                                │  ║         ║ LOOP  │  for every header                                   │ ║            ║                    
                          │                      │ │                           │                                │  ║         ╟───────┘             │ │                                     │ ║            ║                    
                          │                      │ │                           │                                │  ║         ║                     │ │────┐                                │ ║            ║                    
                          │                      │ │                           │                                │  ║         ║                     │ │    │ record.addHeader(name, value)  │ ║            ║                    
                          │                      │ │                           │                                │  ║         ║                     │ │<───┘                                │ ║            ║                    
                          │                      │ │                           │                                │  ║         ╚═════════════════════╪═╪═════════════════════════════════════╪═╝            ║                    
                          │                      │ │                           │                                │  ╚═══════════════════════════════╪═╪═════════════════════════════════════╪══════════════╝                    
                          │                      │ │                           │                                │                                  └┬┘                                     │                                   
                          │                      │ │                           │                records         │                                   │                                      │                                   
                          │                      │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │                      │ │                           │                       sendMessages(producer, records)              │                                     ┌┴┐                                  
                          │                      │ │ ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────>│ │                                  
                          │                      │ │                           │                                │                                   │                                     │ │                                  
                          │                      │ │                           │                                │                                   │                                     │ │                                  
                          │                      │ │                           │                                │                                   │                   ╔═══════╤═════════╪═╪═════════════════════════════════╗
                          │                      │ │                           │                                │                                   │                   ║ LOOP  │  for message in records                     ║
                          │                      │ │                           │                                │                                   │                   ╟───────┘         │ │                                 ║
                          │                      │ │                           │                                │                                   │                   ║                 │ │────┐                            ║
                          │                      │ │                           │                                │                                   │                   ║                 │ │    │ producer.write(message)    ║
                          │                      │ │                           │                                │                                   │                   ║                 │ │<───┘                            ║
                          │                      │ │                           │                                │                                   │                   ╚═════════════════╪═╪═════════════════════════════════╝
                          │                      │ │                           │                                │                                   │                                     └┬┘                                  
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │                      │ │ <─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─│                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │         true         │ │                           │                                │                                   │                                      │                                   
                          │ <────────────────────│ │                           │                                │                                   │                                      │                                   
                          │                      │ │                           │                                │                                   │                                      │                                   
                          │                      └┬┘                           │                                │                                   │                                      │