Kafka Schema Registry – Broker: Der Broker kann den Datensatz nicht validieren

In diesem Artikel stellt Ihnen der PHP-Editor Xiaoxin ein wichtiges Konzept in der Kafka-Architektur-Registrierung vor: Proxy. In Kafka ist der Broker eine Kernkomponente, die für die Verwaltung und Verarbeitung des Nachrichtenflusses verantwortlich ist. Allerdings kann der Proxy keine Datensätze validieren, was bedeutet, dass ein einmal an den Proxy geschriebener Datensatz weder validiert noch geändert werden kann. Diese Funktion kann sich auf bestimmte Nutzungsszenarien und die Sicherheit auswirken. Daher müssen Sie bei der Verwendung von Kafka darauf achten. Als nächstes erklären wir im Detail, warum der Agent den Datensatz nicht überprüfen kann und welche Probleme dabei auftreten können.

Frageninhalt

Ich validiere das Schema mithilfe der Kafka-Schema-Registrierung. Das Problem ist, dass ich immer noch die Fehlermeldung erhalte, obwohl ich das richtige Schema eingegeben habe Broker: Broker failed to verify record.

confluence.value.schema.validation ist auf true gesetzt, damit das Schema des Werts auf der aktuellen Agentenebene überprüft werden kann.

Das von mir eingerichtete Schema und die von mir gesendeten Daten lauten wie folgt.

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "additionalProperties": false,
  "description": "Sample schema to help you get started.",
  "properties": {
    "test": {
      "type": "string"
    }
  },
  "title": "schema_test",
  "type": "object"
}

{"test": "test1"}

Außerdem verwende ich go zum Senden von Daten und der Code für die Daten lautet wie folgt.

// main

func main() {
    kafka.ProduceData("schema_test", []byte(`{"test": "test1"}`))
}

<code>// kafka
func ProduceData(topic string, data []byte) {

    conf := ReadConfig()

    p, err := kafka.NewProducer(&conf)

    if err != nil {
        fmt.Printf("Failed to create producer: %s", err)
        os.Exit(1)
    }

    defer p.Close()

    // Go-routine to handle message delivery reports and
    // possibly other event types (errors, stats, etc)
    go func() {
        for e := range p.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    fmt.Printf("Failed to deliver message: %v\n", ev.TopicPartition)
                } else {
                    fmt.Printf("Produced event to topic %s: key = %-10s value = %s\n",
                        *ev.TopicPartition.Topic, string(ev.Key), string(ev.Value))
                }
            }
        }
    }()

    p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
        Value:          data,
    }, nil)

    // Wait for all messages to be delivered
    p.Flush(15 * 1000)
}
</code>

Richtige Antwort

Es scheint ein Missverständnis darüber zu bestehen, wie Broker Daten tatsächlich überprüfen. Es funktioniert wie erwartet. Sie benötigen eine Schema-ID. Sie senden lediglich einfaches JSON zum Thema ohne die ID. Das Schema in der Registrierung spielt keine Rolle, nur die ID.

Aus der Dokumentation

Genauer gesagt ist das Muster, das Sie der Registrierung hinzufügen, nur eine von vielen „Versionen“, die möglicherweise für das Thema vorhanden sind (z. B. topic-value). Jede Version hat eine eindeutige ID. Bei der Authentifizierung wird nicht nur die neueste Version verwendet; die ID wird auch auf der Clientseite kodiert.

Sehen Sie sich das Confluence-Beispiel mit der JSON-Schemagenerierung an (das selbst die Datensatzvalidierung durchführen sollte).

https://github.com/confluenceinc/confluence-kafka-go/blob/master/examples/json_ Producer_example/json_ Producer_example.go

Die Validierung auf Brokerseite dient lediglich dazu, falsch serialisierte Daten oder „Giftpillen“ zu verhindern, wie Sie es jetzt tun.

Das obige ist der detaillierte Inhalt vonKafka Schema Registry – Broker: Der Broker kann den Datensatz nicht validieren. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!