Блог

Я развернул кластер Kafka и кластер Kafka Connect в kubernetes, используя Strimzi и AKS. И я хотел начать чтение из RSS-ресурсов для своего кластера Kafka, поэтому я создал экземпляр соединителя «org.kaliy.kafka.connect.rss.RssSourceConnector», который считывает из определенного RSS-канала, учитывая URL-адрес, и записывает в определенную тему. Но все мое намерение состоит в том, чтобы в конечном итоге создать кластер Kafka Connect, способный управлять множеством внешних запросов новых RSS для чтения; и вот тут-то и возникают все мои сомнения:

• Должен ли я создать экземпляр соединителя Kaliy RSS для каждого канала RSS? Или было бы лучше реализовать мой собственный соединитель, чтобы я создавал только один его экземпляр и каждый раз, когда я хочу прочитать новый канал RSS, я бы создавал новую задачу в соединителе?
• Кто должен отвечать за обеспечение желаемого состояния кластера Kafka Connect? Я имею в виду, что если соединитель (в случае 1 RSS-канала: 1 экземпляр соединителя) перестал работать, кто должен попытаться запустить его снова? Внешний клиент через API Kafka Connect REST? Сам Kubernetes?

Прямо сейчас, я думаю, что мой лучший вариант — полагаться на API Kafka Connect REST, который делает внешнего клиента ответственным за управление состоянием набора соединителей, но я не знаю, были ли они разработаны для получения большого количества запросов, как это было бы в случае. Возможно, их можно было бы масштабировать, предоставив несколько прослушивателей в конфигурации API REST Kafka Connect, но я не знаю. Большое спасибо!

Одним из основных преимуществ использования Kafka Connect является использование подхода, основанного на настройке, поэтому вы потеряете это, внедрив свой собственный соединитель. На мой взгляд, лучшая стратегия — иметь один экземпляр соединителя для каждого RSS-канала. Уменьшение количества экземпляров может иметь смысл при наличии единой системы источников данных, чтобы избежать ее перегрузки.

Используя оператор Strimzi, кластер Kafka Connect будет контролироваться, и при необходимости он попытается восстановить желаемое состояние кластера. Это не включает экземпляры отдельных соединителей и их задачи, но вы можете использовать API K8s для мониторинга состояния пользовательского ресурса (CR) соединителя вместо REST API.

Пример:

 $ kubetctl get kafkaconnector amq-sink -o yaml
apiVersion: kafka.strimzi.io/v1alpha1
kind: KafkaConnector
# ...
status:
  conditions:
  - lastTransitionTime: "2020-12-07T10:30:28.349Z"
    status: "True"
    type: Ready
  connectorStatus:
    connector:
      state: RUNNING
      worker_id: 10.116.0.66:8083
    name: amq-sink
    tasks:
    - id: 0
      state: RUNNING
      worker_id: 10.116.0.66:8083
    type: sink
  observedGeneration: 1