Schema-Registry : Producer

목차

[주요 Configuration]

#schema-registry url
spring.kafka.producer.properties.schema.registry.url: http;://localhost:8081
 
# schema 자동등록 여부 (default: true. 운영에서는 false 권고)
spring.kafka.producer.properties.auto.register.schemas: false 
 
# 최신버전의 schema만 사용할지 여부 (default: false)
spring.kafka.producer.properties.use.latest.version: false  
 
# key-serializer
spring.kafka.producer.properties.key-serializer: org.apache.kafka.common.serialization.StringSerializer
 
# value-serializer
spring.kafka.producer.properties.value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer

 

• serializer
  • serializer에 "io.confluent.kafka.serializers.KafkaAvroSerializer"를 지정하는 경우 Application이 구동될 때 schema-registry.url 같은 값을 자동으로 찾는다. 없을 경우 오류가 발생한다. 
• auto.register.schemas
  • 스키마가 등록되어 있지 않을 경우 자동으로 등록할지 여부 
  • 운영환경에서는 false가 권고된다.
• use.latest.version
  • 최신버전의 schema만 사용할지 여부 

 

[Configuration 값에 따른 동작 확인]

serializer

serializer에 io.confluent.kafka.serializers.KafkaAvroSerializer 설정 후 schema.registry.url를 지정하지 않았을 때 오류발생하는지 확인

Action	확인사항	확인결과	비고
key, value-serializer에 StringSerializer로 지정	Application 구동시 Schema.Registry.Url로 찾는지 확인	Schema.Registry.Url 값이 없으니 오류가 발생

 

auto.register.schemas

값에 따른 동작 확인

샘플 schema

{
  "fields": [
    {
      "name": "name",
      "type": {
        "avro.java.string": "String",
        "type": "string"
      }
    },
    {
      "name": "age",
      "type": "int"
    }
  ],
  "name": "MemberBackward",
  "namespace": "com.ebaykorea.schema.registry.tester.producer.avro",
  "type": "record"
}

 

true

Action	확인사항	확인결과	비고
토픽에 schema 없는 상태 확인
spring.kafka.producer.properties.auto.register.schemas true로 지정	Application 구동 시 auto.register.shemas = true인지 확인	확인
토픽 메시지 생성	메시지에 대한 schema가 없으니 자동으로 생성하는지 확인	확인

false

Action	확인사항	확인결과	비고
토픽에 schema 없는 상태 확인
spring.kafka.producer.properties.auto.register.schemas false로 지정	Application 구동 시 auto.register.shemas = false인지 확인	확인
토픽 메시지 생성	메시지에 대한 schema가 없으니 Producer에서 오류 발생	확인

use.latest,version

값에 따른 동작 확인

 

초기 schema

 

{
  "fields": [
    {
      "name": "name",
      "type": {
        "avro.java.string": "String",
        "type": "string"
      }
    },
    {
      "name": "age",
      "type": "int"
    }
  ],
  "name": "MemberBackward",
  "namespace": "com.ebaykorea.schema.registry.tester.producer.avro",
  "type": "record"
}

 

변경될 schema (address field 추가)

{
  "fields": [
    {
      "name": "name",
      "type": {
        "avro.java.string": "String",
        "type": "string"
      }
    },
    {
      "name": "age",
      "type": "int"
    },
    {
      "default": "busan",
      "name": "address",
      "type": {
        "avro.java.string": "String",
        "type": "string"
      }
    }
  ],
  "name": "MemberBackward",
  "namespace": "com.ebaykorea.schema.registry.tester.producer.avro",
  "type": "record"
}

 

Action	확인사항	확인결과	비고
producer에서 use.latest.version = true로 지정	Producing 성공 확인
Producer에서 기존 Schema로 재시작	Producing 실패 확인	실패	Producer에서 Schema 캐싱이 다시 되었는지 실패한다.
Producer에서는 기존 schema로 계속 구동	Producing 실패 확인	실패한다고 생각했으나 계속 정상적으로 동작한다.	로컬 캐시와 연관된 것 같다.
topic에서 schema 변경 (address 필드 추가)			정책을 Backward로 했기 때문에 default 값을 넣었다.
topic에서 schema를 name,age만 있는 schema로 세팅