개요

전 포스티에서 connector 구현하던 중 애매하게 끝났던 부분이 있었습니다. strimzi 에서 KafkaConnect를 사용할 이미지를 커스텀하는 방법입니다.

해당 내용과 관련하여 정리해보고 실제로 connector 를 어떻게 구현했는지 공유해보겠습니다.

customize the strimzi connect image

Building a new container image with connector plugins from the Kafka Connect base image

위 내용을 참고하시면 원하는 플러그인을 추가하여 새로운 이미지를 만들 수 있습니다. 이때는 다양한 플러그인을 사용할 수 있습니다.

대표적으로 사용할 수 있는 플러그인을 접할 수 있는 공간은 아래와 같이 있습니다.
confluent.io hub
debezium
camel kafka

제일 많이 접하고 나름대로 connect 관련 제일 공신력있는 3개를 선택했습니다. 적용하는 방법은 문서를 통해서 확인하시는게 제일 좋습니다.

사실 저는 이미지를 새로 만들어서 사용하지 않았습니다.

관리하는게 귀찮습니다.

• 버전
• 빌드 스크립트
• 개인 registry

그래서 최대한 building a new container 방식을 추구하고 하나의 yaml 에서 모두 관리하고 싶었습니다. -> 이것도 매번 빌드 되는 것이 아니냐 이렇게 생각하실 수 있겠지만 스크립트만 나눠서 관리할 수 있을 것 같습니다.

여하튼 돌아와서 그래서 저는 debezium - Jdbc 를 사용하기로 결정했습니다. 사실 앞 방법을 사용할 위험도가 있는건 사실입니다. 전 포스팅에서도 말했지만 Incubating 단계이기 때문입니다.

그럼에도 불구하고 선택한 이유는 여러가지가 있습니다.

1. 지금 제가 kafka 를 해보는 이유는 🥟맛 보기 용도입니다.
2. 필드 변경이 걱정되었지만 큰 타격이 있을정도로 변하지 않을 것 같습니다. (개인 추측)

이정도 입니다. 그래서 비교적 간단하게 만들었습니다.

manifest s

KafkaConnect

사실 저는 글에 예제코드를 적는걸 별로 좋아하지 않습니다. 이게 문서에 나와있는걸 다시 한 번 적는 느낌이 나서말이죠. 이번에 간결하게 적고 조금 주의하시면 좋을 듯한 내용이 있어 적어보도록 하겠습니다.

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 1
  bootstrapServers: my-cluster-kafka-bootstrap:9092 # Z. ca 가 존재한다면 9093 사용하세요. 이걸 추천드립니다.
  image: example.registry/kafka/kafkaconnect:latest # tagert registry 
  config:
    config.providers: secrets
    config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
    group.id: connect-cluster # A. unique 해야 합니다.
    offset.storage.topic: connect-cluster-offsets #B. 문서에서 확인할 수 있습니다.
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
      # key.converter: # C. 아직은 사용할 필요가 없어서 적용하지 않았습니다.
      # key.converter.schema.registry.url: 
      # value.converter: 
      # value.converter.schema.registry.url: 
  template:
    pod:
      imagePullSecrets:
        - name: example-secret #D. private registry 일 경우
  build:
    output:
      type: docker
      image: example.registry/kafka/kafkaconnect:latest # C. tagert registry 
      pushSecret: example-secret
    plugins:
      - name: debezium-mysql-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.2.0.Final/debezium-connector-mysql-2.2.0.Final-plugin.tar.gz
      - name: debezium-connector-jdbc
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-jdbc/2.2.1.Final/debezium-connector-jdbc-2.2.1.Final-plugin.tar.gz

Z. bootstrapServers: my-cluster-kafka-bootstrap:9092

이 글을 보고 처음부터 debezium 으로 시작할 수도 있을 것 같았습니다. 저는 helm 으로 구성했으며 service는 아래와 같이 구성되어 있습니다.
tcp-ctrlplane:9090►0 tcp-replication:9091►0 tcp-clients:9092►0 tcp-clientstls:9093►0
9092를 사용한 이유는 9093 썻다가 ca 없는 분들은 에러나실 수도 있을 것 같아서 일단 9092를 사용했습니다.
ca가 존재한다면 tls 설정과 9093 사용하는 것이 보다 안전합니다.

A. group.id: connect-cluster
거의 그럴일은 없지만 connect cluster 를 여러 클러스터로 구성하는 경우는 unique한 값들에 신경써야합니다. 가령 예를 들면 위 이름이 같은 클러스터가 2개 이상이라면 connector 입장에서 혼돈이 생길 수 있습니다. 때문에 plugin은 정상적으로 됐지만 connector가 500을 뱉을 수도 있습니다.

B. offset.storage ...등등
해당 내용은 문서에서 확인할 수 있습니다. 만약에 중간에 object 를 다시 생성하거나 그런 경우 설정이 필요할 수도 있을 것 같습니다. (사실 이건 저도 세세하게 설정해본 경험은 없습니다.)

C. key, value converter
convert 역시 다양한 종류가 있습니다. 그중에서도 confluent Schema Registry 를 많이 사용하는 것 같습니다. 처음 본다면 요거 Schema Registry Overview 보시면 개념확립하시는데 많은 도움이 될겁니다.

(converter 의 경우는 저도 개념만 알고 있습니다. debezium 의 connector 들을 사용하면 현재는 converter의 크게 의존할 필요는 없다고 생각했습니다. 하지만 굳이 말씀드리는 이유는 만약 다양한 connector, log 등을 사용하면서 나중에 서비스단계에서 작업들을 줄이기위한 방법으로 요긴하게 사용할 수 있을 것 같다는 생각을 많이해서 추천드립니다. 저도 나중에 해볼 생각입니다. 이것도 기회가 된다면 글로 작성하고 싶네요 !)

D. private registry 라면
귀찮은 작업이긴하지만 그래도 그나마 ㅎㅎㅎㅎ...
docker image 를 push 하기 위해서 pushSecret: example-secret 를 통해서 인증작업을 수행해야합니다. registry가 public하여 모든 사람들에게 열려있는 상태라면 여기까지만 작업하셔도 됩니다.
하지만 pull 하기 위해 인증이 필요하다면 secret 파일이 필요하며 -> template.pod.imagePullSecrets:[] 사용해주시면 됩니다. 이건 strimzi 에서 못봐서 직접 찾았던 기억이 있습니다. (이것때문에 CRD 구경도 해보고 좋았습니다.)

정상적으로 배포된 모습과 플러그인내용들도 확인할 수 있습니다.

connector

connector는 모두 debezium 을 사용했습니다. 때문에 문서 보시는게 더 정확합니다.

하나 말하자면 두 connector 에서 필수로 존재해야하는 설정들이 있습니다. 또 하나의 설정이 다른 필드에 영향을 주는 경우도 있으니 주의하고 , 안되면 log 잘 보시면 확인할 수 있습니다.

로그처리를 정말 잘 해둔 debezium 입니다 :)

하나만 예를 들면
spec.config.schema.evolution 필드의 속성값 설정에 따라 spec.config.primary.key.fields 필드의 영향을 줍니다. 이점은 documentation 을 참고해주세요 !

connetors 역시 정상적으로 배포된 것과 어떤 클래스를 사용하고 있는지 확ㅇ니할 수 있습니다.

🏆마치며

앞으로 kafka 글을 계속해서 작성할 예정입니다.
최근에 정말 많은 생각을 합니다만 쿠버네티스는 De facto 로 혼자서 예상해보고 있습니다. 주니어이긴하지만 .. IDC, bare metal 경험을 해보니 보면 볼 수록 놀라는게 .. k8s 입니다.

더해서 최근에 kafka 를 보면서 규모가 있는 기업이라면 꼭 필요한 기술처럼 생각되고 어떤 서비스와 이질적이지 않은 느낌의 기술이라는 것을 느끼면서 재미있었습니다.

한국에서 아직은 strimzi , 다양한 connector , 관련 주변 기술들의 내용이 부족한 것을 보고 조금 아쉬워서 이렇게라도 조금 정보가 생기길 바랬습니다.

또 말이 길어졌습니다. 여기까지 입니다 😁