Real Time API in the context of Apache Kafka

Hello everyone. Already in December, OTUS will launch a new stream of Software Architect course . In anticipation of the start of the course, I want to share with you the translation of an interesting article. I also suggest watching a demo lesson on the topic: "Idempotency and commutability of API in queues and HTTP" .

---

One of the tricky questions that we constantly face when designing applications and systems in general is how to effectively organize the exchange of information between components, while maintaining the flexibility to change interfaces without unduly affecting other parts of the system. The more specific and optimized an interface is, the more likely it is to be so situational that it will require a complete rewrite to change it. And vice versa; Generics can be quite adaptive and widely supported, but alas, at the expense of performance.

Events offer a Goldilocks-style approach in which real-time APIs can be used as the basis for applications that are flexible yet high performing; loosely coupled but effective.

Events can be thought of as building blocks for many other data structures. Typically, they record the fact that something happened and the point in time at which it happened. An event can capture this information at various levels of detail, from a simple notification to a detailed event describing the full state of what happened.

(state) - , NoSQL . , , - - . , (consumers) - , . (producers) , , .

, , , , . , . (payload) , , . :

• : userLogin
  
  
  
  

  
  
  
  
  

  • : zbeeblebrox



2020-08-17 16:26:39 BST
    
    
    
    

    
    
    
    
    

• : CarParked
  
  
  
  

  
  
  
  
  

  • : A42 XYZ



2020-08-17 16:36:27



X42
    
    
    
    

    
    
    
    
    

• : orderPlaced
  
  
  
  

  
  
  
  
  

  • :
    
    
    
    
    
    
    
    £2.25



2020-08-17 16:35:41 BST
    
    
    
    

    
    
    
    
    
    
    
    
    
    

(, , ), (, , , , ).

, , , , - Apache Kafka®. Kafka - , :

• Pub/Sub

  
  
  
  
  

  • () () , / .

    
    
    
    
    

• 
  
  
  
  

  • , .

    
    
    
    
    

• 
  
  
  
  

  • .

    
    
    
    
    

Kafka . , , , , , . , , , , NoSQL .

API-, Apache Kafka, , , .

API

, Kafka, , , , , , ( - ). , Kafka - . , , - Kafka (topic - Kafka, - ), .

Kafka . Kafka Go:

package main

import (
    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)

func main() {

    topic := "test_topic"
    p, _ := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092"})
    defer p.Close()

    p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic,
            Partition: 0},
        Value: []byte("Hello world")}, nil)

}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Kafka , , , , , ( ).

, Kafka, . pub/sub, , , . Kafka , , , A/B-, , . , , . , RabbitMQ, ActiveMQ, .

package main

import (
    "fmt"

    "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka"
)

func main() {

    topic := "test_topic"

    cm := kafka.ConfigMap{
        "bootstrap.servers":        "localhost:9092",
        "go.events.channel.enable": true,
        "group.id":                 "rmoff_01"}

    c, _ := kafka.NewConsumer(&cm)
    defer c.Close()
    c.Subscribe(topic, nil)

    for {
        select {
        case ev := <-c.Events():
            switch ev.(type) {

            case *kafka.Message:
                km := ev.(*kafka.Message)
                fmt.Printf("✅ Message '%v' received from topic '%v'\n", string(km.Value), string(*km.TopicPartition.Topic))
            }
        }
    }

}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Kafka, (Consumer Group). . -, Kafka , , , . -, - , , . Kafka , (, ).

, - . Kafka Connect API, .

Producer Consumer API Java, C/C++, Go, Python, Node.js . , HTTP Kafka? REST Proxy.

REST API Apache Kafka

, . , , , :

{
    "name": "NCP Sheffield",
    "space": "A42",
    "occupied": true
}
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Kafka, . Kafka Confluent REST Proxy - REST-:

curl -X POST \
     -H "Content-Type: application/vnd.kafka.json.v2+json" \
     -H "Accept: application/vnd.kafka.v2+json" \
     --data '{"records":[{"value":{ "name": "NCP Sheffield", "space": "A42", "occupied": true }}]}' \
     "http://localhost:8082/topics/carpark"
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, Consumer API, , REST-. Consumer API, , REST API, Consumer Group, (subscription). , REST API , :

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \
      --data '{"name": "rmoff_consumer", "format": "json", "auto.offset.reset": "earliest"}' \
      http://localhost:8082/consumers/rmoff_consumer

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" --data '{"topics":["carpark"]}' \
 http://localhost:8082/consumers/rmoff_consumer/instances/rmoff_consumer/subscription
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

:

curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \
      http://localhost:8082/consumers/rmoff_consumer/instances/rmoff_consumer/records
[
    {
        "topic": "carpark",
        "key": null,
        "value": {
            "name": "Sheffield NCP",
            "space": "A42",
            "occupied": true
        },
        "partition": 0,
        "offset": 0
    }
]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, . , REST-.

, Kafka. , pub/sub. - , , ? , , .

,

Apache Kafka pub/sub - iPhone . , … . Apache Kafka Kafka Streams API. Java Kafka . Kafka Streams, , Walmart, Ticketmaster Bloomberg, ksqlDB.

ksqlDB - , . API SQL Kafka. ksqlDB , , - .

ksqlDB :

CREATE STREAM CARPARK_EVENTS (NAME     VARCHAR,
                              SPACE    VARCHAR,
                              OCCUPIED BOOLEAN)
                        WITH (KAFKA_TOPIC='carpark',
                              VALUE_FORMAT='JSON');
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ksqlDB , , . , . , , , :

SELECT TIMESTAMPTOSTRING(ROWTIME,'yyyy-MM-dd HH:mm:ss') AS EVENT_TS,
       SPACE
  FROM CARPARK_EVENTS
 WHERE NAME='Sheffield NCP'
   AND OCCUPIED=false
  EMIT CHANGES;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

SQL-, , , , ( EMIT CHANGES). , push-, , , . ksqlDB pull- ( ), , , . , ksqlDB , , .

ksqlDB REST API, SQL :

curl --http2 'http://localhost:8088/query-stream' \
     --data-raw '{"sql":"SELECT TIMESTAMPTOSTRING(ROWTIME,'\''yyyy-MM-dd HH:mm:ss'\'') AS EVENT_TS, SPACE FROM CARPARK_EVENTS WHERE NAME='\''Sheffield NCP'\'' and OCCUPIED=false EMIT CHANGES;"}'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, :

{"queryId":"383894a7-05ee-4ec8-bb3b-c5ad39811539","columnNames":["EVENT_TS","SPACE"],"columnTypes":["STRING","STRING"]}
…
["2020-08-05 16:02:33","A42"]
…
…
…
["2020-08-05 16:07:31","D72"]
…
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ksqlDB . SELECT



CREATE STREAM streamname AS



Kafka. , ksqlDB , , .. , Kafka. ksqlDB , , :

CREATE STREAM CARPARKS AS
    SELECT E.NAME AS NAME, E.SPACE,
           R.LOCATION, R.CAPACITY,
           E.OCCUPIED,
           CASE
               WHEN OCCUPIED=TRUE THEN 1
               ELSE -1
           END AS OCCUPIED_IND
    FROM   CARPARK_EVENTS E
           INNER JOIN
           CARPARK_REFERENCE R
           ON E.NAME = R.NAME;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, CASE , . CREATE STREAM Kafka, :

+----------------+-------+----------+----------------------------+----------+--------------+
|NAME            |SPACE  |OCCUPIED  |LOCATION                    |CAPACITY  |OCCUPIED_IND  |
+----------------+-------+----------+----------------------------+----------+--------------+
|Sheffield NCP   |E48    |true      |{LAT=53.4265964, LON=-1.8426|1000      |1             |
|                |       |          |386}                        |          |              |
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , ksqlDB . , SQL, :

CREATE TABLE CARPARK_SPACES AS
    SELECT NAME,
           SUM(OCCUPIED_IND) AS OCCUPIED_SPACES
        FROM CARPARKS
        GROUP BY NAME;
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

ksqlDB REST API:

curl --http2 'http://localhost:8088/query-stream' \
     --data-raw '{"sql":"SELECT OCCUPIED_SPACES FROM CARPARK_SPACES WHERE NAME='\''Birmingham NCP'\'';"}'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , ( "pull- ", " push- ") , :

{"queryId":null,"columnNames":["OCCUPIED_SPACES"],"columnTypes":["INTEGER"]}
[30]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, -

curl --http2 'http://localhost:8088/query-stream' \
     --data-raw '{"sql":"SELECT OCCUPIED_SPACES FROM CARPARK_SPACES WHERE NAME='\''Birmingham NCP'\'';"}'
{"queryId":null,"columnNames":["OCCUPIED_SPACES"],"columnTypes":["INTEGER"]}
[29]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

Java ksqlDB Python Go.

Apache Kafka , , , ( ), .

, , , , , :

• 
  
  
  
  

• 
  
  
  
  

• 
  
  
  
  

Apache Kafka Connect API, , Kafka. , Kafka S3 :

curl -i -X PUT -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/sink-s3/config \
    -d ' {
        "connector.class": "io.confluent.connect.s3.S3SinkConnector",
        "topics": "carpark",
        "s3.bucket.name": "rmoff-carparks",
        "s3.region": "us-west-2",
        "flush.size": "1024",
        "storage.class": "io.confluent.connect.s3.storage.S3Storage",
        "format.class": "io.confluent.connect.s3.format.json.JsonFormat"
        }'
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
    

, , , S3. . , Snowflake, Kafka Connect; . Kafka Connect Kafka. , CARPARK_REFERENCE



, ksqlDB , (CDC - change data capture) , .

Apache Kafka , . , , , , , .

API Kafka , ksqlDB, , / . API , REST.

Apache Kafka, developer.confluent.io. Confluent Platform - Apache Kafka, , . , Confluent Cloud.  GitHub Docker Compose , . - , Kafka, « - ».

---

- : " API HTTP".

---