如何操作Strimzi的KafkaConnector资源实现Kafka连接器管理

初始方案（直接调用Kafka Connect REST API） #

最开始尝试直接通过Kafka Connect REST API管理连接器，暂停功能的简化实现代码如下：

def pause(): Unit = {
      logger.info(s"pause() Triggered")
      val response = HttpClient.newHttpClient.send({
        HttpRequest
          .newBuilder(URI.create(config.connectUrl + s"/connectors/${config.connectorName}/pause"))
          .PUT(BodyPublishers.noBody)
          .timeout(Duration.ofMillis(config.timeout.toMillis.toInt))
          .build()
      }, BodyHandlers.ofString)
      if (response.statusCode() != HTTPStatus.Accepted) {
        throw new Exception(s"Could not pause connector: ${response.body}")
      }
    }

直接调用API的问题 #

使用KafkaConnector作为托管资源时，连接器Operator会将Kubernetes上的KafkaConnector自定义资源作为唯一可信源，直接通过REST API执行的暂停等手动修改会被Cluster Operator自动回滚，因此该方案无法在Operator托管场景下使用。

基于Kubernetes CRD的最终实现方案 #

通过直接修改KafkaConnector CRD资源的方式实现连接器管理，在@Jakub的协助下完成了适配实现，支持暂停/创建/删除等全生命周期管理，同时增加了状态校验逻辑保证操作生效，完整代码如下：

class KubernetesService(config: Configuration) extends StrictLogging {

  private[this] val client = new DefaultKubernetesClient(Config.autoConfigure(config.connectorContext))

  def setPause(pause: Boolean): Unit = {
    logger.info(s"[KubernetesService] - setPause($pause) Triggered")

    val connector = getConnector()
    connector.getSpec.setPause(pause)
    Crds.kafkaConnectorOperation(client).inNamespace(config.connectorNamespace).withName(config.connectorName).replace(connector)

    Crds.kafkaConnectorOperation(client)
      .inNamespace(config.connectorNamespace)
      .withName(config.connectorName)
      .waitUntilCondition(connector => {
        connector != null &&
          connector.getSpec.getPause == pause && {
          val desiredState = if (pause) "Paused" else "Running"
          connector.getStatus.getConditions.stream().anyMatch(_.getType.equalsIgnoreCase(desiredState))
        }
      }, config.timeout.toMillis, TimeUnit.MILLISECONDS)
  }

  def delete(): Unit = {
    logger.info(s"[KubernetesService] - delete() Triggered")

    Crds.kafkaConnectorOperation(client).inNamespace(config.connectorNamespace).withName(config.connectorName).delete

    Crds.kafkaConnectorOperation(client)
      .inNamespace(config.connectorNamespace)
      .withName(config.connectorName)
      .waitUntilCondition(_ == null, config.timeout.toMillis, TimeUnit.MILLISECONDS)
  }

  def create(oldKafkaConnect: KafkaConnector): Unit = {
    logger.info(s"[KubernetesService] - create(${oldKafkaConnect.getMetadata}) Triggered")

    Crds.kafkaConnectorOperation(client).inNamespace(config.connectorNamespace).withName(config.connectorName).create(oldKafkaConnect)

    Crds.kafkaConnectorOperation(client)
      .inNamespace(config.connectorNamespace)
      .withName(config.connectorName)
      .waitUntilCondition(connector => {
          connector != null &&
          connector.getStatus.getConditions.stream().anyMatch(_.getType.equalsIgnoreCase("Running"))
      }, config.timeout.toMillis, TimeUnit.MILLISECONDS)
  }

  def getConnector(): KafkaConnector = {
    logger.info(s"[KubernetesService] - getConnector() Triggered")
    Try {
      Crds.kafkaConnectorOperation(client).inNamespace(config.connectorNamespace).withName(config.connectorName).get
    } match {
      case Success(connector) => connector
      case Failure(_: NullPointerException) => throw new NullPointerException(s"Failure on getConnector(${config.connectorName}) on ns: ${config.connectorNamespace}, context: ${config.connectorContext}")
      case Failure(exception) => throw exception
    }
  }
}

内容的提问来源于stack exchange，提问作者Zvi Mints