kafka-connect pinging the broker almost continuously when using a custome authorizer and principal builder

Question

I am running Kafka in docker using the cp-all-in-one docker-compose file provided by confluent. I have modified some settings to be able to use OAuth for authentication and authorization.

My setup is working correctly so far, but I have an issue where the kafka-connect container is pinging the broker like crazy. It's sending Read requests to the broker almost continuously. I am using a custom authorizer and principal builder class following this authorizer and principal-builder. I needed these for controlling authorization via OAuth scopes. Here's an excerpt from the crazy logs I'm getting.

broker             | [2021-01-12 08:23:16,582] INFO >> Building CustomPrincipal (com.bfm.kafka.security.oauthbearer.CustomPrincipalBuilder)
broker             | [2021-01-12 08:23:16,582] INFO >> Plaintext Authentication Context (com.bfm.kafka.security.oauthbearer.CustomPrincipalBuilder)
broker             | [2021-01-12 08:23:16,583] INFO Starting Authorization. (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,583] INFO Operation request Info: Read (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,583] INFO Resource request Info: Topic:LITERAL:test (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,583] INFO KafkaPrincipal: User:ANONYMOUS (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,583] INFO Client using Plaintext. No authorization is performed (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,642] INFO >> Building CustomPrincipal (com.bfm.kafka.security.oauthbearer.CustomPrincipalBuilder)
....
broker             | [2021-01-12 08:23:16,643] INFO Operation request Info: Read (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,643] INFO Resource request Info: Topic:LITERAL:_kafka-connect-offsets (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
...
broker             | [2021-01-12 08:23:16,675] INFO Operation request Info: Read (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,675] INFO Resource request Info: Topic:LITERAL:_kafka-connect-configs (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,675] INFO KafkaPrincipal: User:ANONYMOUS (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
...
broker             | [2021-01-12 08:23:16,696] INFO Operation request Info: Read (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,696] INFO Resource request Info: Topic:LITERAL:_kafka-connect-status (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,696] INFO KafkaPrincipal: User:ANONYMOUS (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)
broker             | [2021-01-12 08:23:16,696] INFO Client using Plaintext. No authorization is performed (com.bfm.kafka.security.oauthbearer.CustomAuthorizer)

You can see that all of this is happening within 100 milliseconds. And this is happening continuously without any pause. Is this normal behavior for kafka-connect to ping broker continuously or is there a problem with my setup? I am adding relevant docker-compose setups. Some values are not included for brevity.

broker:
    image: confluentinc/cp-kafka:5.5.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - 29092:29092
      - 9092:9092
      - 9093:9093
    environment:
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT,CLIENT:SASL_PLAINTEXT
      KAFKA_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://broker:9092,CLIENT://broker:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092,CLIENT://localhost:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      
      KAFKA_LISTENER_NAME_CLIENT_OAUTHBEARER_SASL_LOGIN_CALLBACK_HANDLER_CLASS: com.bfm.kafka.security.oauthbearer.OAuthAuthenticateLoginCallbackHandler
      KAFKA_LISTENER_NAME_CLIENT_OAUTHBEARER_SASL_SERVER_CALLBACK_HANDLER_CLASS: com.bfm.kafka.security.oauthbearer.OAuthAuthenticateValidatorCallbackHandler
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: OAUTHBEARER
      KAFKA_SASL_ENABLED_MECHANISMS: OAUTHBEARER
      KAFKA_AUTHORIZER_CLASS_NAME: com.bfm.kafka.security.oauthbearer.CustomAuthorizer
      KAFKA_PRINCIPAL_BUILDER_CLASS: com.bfm.kafka.security.oauthbearer.CustomPrincipalBuilder
      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"
      ZOOKEEPER_SASL_ENABLED: "FALSE"
    volumes: 
      - ./kafka-broker/config/kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
      - ./kafka-broker/libs/libkafka.oauthbearer-1.0.0.jar:/usr/share/java/kafka/libkafka.oauthbearer-1.0.0.jar

schema-registry:
    image: confluentinc/cp-schema-registry:5.5.0
    container_name: schema-registry
    depends_on:
      - zookeeper
      - broker
    ports:
      - 8081:8081
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper:2181

kafka-connect:
    image: confluentinc/cp-kafka-connect-base:5.5.0
    container_name: kafka-connect
    depends_on:
      - broker
      - schema-registry
    ports:
      - 8083:8083
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "broker:29092"
      CONNECT_REST_ADVERTISED_HOST_NAME: "kafka-connect"
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: kafka-connect
      CONNECT_CONFIG_STORAGE_TOPIC: _kafka-connect-configs
      CONNECT_OFFSET_STORAGE_TOPIC: _kafka-connect-offsets
      CONNECT_STATUS_STORAGE_TOPIC: _kafka-connect-status
      CONNECT_KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_LOG4J_ROOT_LOGLEVEL: "INFO"
      CONNECT_LOG4J_LOGGERS: "org.apache.kafka.connect.runtime.rest=WARN,org.reflections=ERROR"
      CONNECT_LOG4J_APPENDER_STDOUT_LAYOUT_CONVERSIONPATTERN: "[%d] %p %X{connector.context}%m (%c:%L)%n"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "1"
      CONNECT_PLUGIN_PATH: '/usr/share/java,/usr/share/confluent-hub-components/,/connectors/'
    volumes:
      - $PWD/connectors:/connectors

Is this normal behavior for kafka-connect or am is it due to my setup? Any help/pointer is greatly appreciated. Thanks.

Answer 1

等待大神答复