schema registry not working validation

[kgw7401]

Description

I am currently using the schema registry to validate the schema of data being produced. However, even though I produce data in a form different from the registered schema, the data is sent successfully. I am using Go.

The schema i set and the data i send are as follows.

{
    "properties": {
      "device_id": {
        "type": "string"
      },
      "event_name": {
        "type": "string"
      },
      "platform": {
        "enum": [
          "android",
          "ios",
          "web"
        ],
        "type": "string"
      },
      "user_id": {
        "type": "string"
      }
    },
    "required": [
      "event_name",
      "user_id",
      "device_id",
      "platform"
    ],
    "title": "view_home",
    "type": "object"
  }

{
  "name": "First user",
  "favorite_number": "42",
  "favorite_color": "blue"
}

Additionally, I am sending data using go and the code for the data is as follows.

func ProduceData(topic string, data []byte) {

	var d interface{}

	switch topic {
	case "view_home":
		d = ViewHome{}
		json.Unmarshal(data, &d)
	case "view_searchResult":
		d = ViewSearchResult{}
		json.Unmarshal(data, &d)
	}

	conf := ReadConfig()

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

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

	defer p.Close()

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

	fmt.Printf("Created Producer %v\n", p)

	client, err := schemaregistry.NewClient(schemaregistry.NewConfig("http://localhost:8081"))

	if err != nil {
		fmt.Printf("Failed to create schema registry client: %s\n", err)
		os.Exit(1)
	}

	serdeConfig := jsonschema.NewSerializerConfig()
	serdeConfig.AutoRegisterSchemas = false
	serdeConfig.UseLatestVersion = true
	serdeConfig.EnableValidation = true

	ser, err := jsonschema.NewSerializer(client, serde.ValueSerde, serdeConfig)

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

	// Optional delivery channel, if not specified the Producer object's
	// .Events channel is used.
	deliveryChan := make(chan kafka.Event)
	value := User{
		Name:           "First user",
		FavoriteNumber: "42",
		FavoriteColor:  "blue",
	}
	payload, err := ser.Serialize(topic, &value)
	if err != nil {
		fmt.Printf("Failed to serialize payload: %s\n", err)
		os.Exit(1)
	}
	err = p.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          payload,
	}, deliveryChan)
	if err != nil {
		fmt.Printf("Produce failed: %v\n", err)
		os.Exit(1)
	}

	e := <-deliveryChan
	m := e.(*kafka.Message)

	if m.TopicPartition.Error != nil {
		fmt.Printf("Delivery failed: %v\n", m.TopicPartition.Error)
	} else {
		fmt.Printf("Delivered message to topic %s [%d] at offset %v\n",
			*m.TopicPartition.Topic, m.TopicPartition.Partition, m.TopicPartition.Offset)
	}

	close(deliveryChan)
}

As you can see from the code above, even though the registered schema and the actual data being sent are completely different, the data is serialized and sent normally. The settings for the serialization process are as follows.

serdeConfig := jsonschema.NewSerializerConfig()
serdeConfig.AutoRegisterSchemas = false
serdeConfig.UseLatestVersion = true
serdeConfig.EnableValidation = true

What could be the problem?

Checklist

Please provide the following information:

• confluent-kafka-go and librdkafka version (LibraryVersion()):
• Apache Kafka broker version:
• Client configuration: ConfigMap{...}
• Operating system:
• Provide client logs (with "debug": ".." as necessary)
• Provide broker log excerpts
• Critical issue

[rayokota]

This is a known issue, related to #984