无法使用 akka kafka kerberos 配置在 kafka 消费者端接收 Kafka 消息答案

【问题标题】：Unable To Receive Kafka Messages at kafka consumer side with akka kafka kerberos configuration无法使用 akka kafka kerberos 配置在 kafka 消费者端接收 Kafka 消息
【发布时间】：2017-06-07 15:39:27
【问题描述】：

我正在使用 akka、kafka、scala 开发应用程序，并且我的应用程序在我的本地系统上运行良好，但是每当我在启用了 kerberos 安全性的集群上部署它时，我都无法在 kafka 消费者端接收消息。出于测试目的，我正在从 kafka 命令行运行生产者并从命令行发送消息。如果我的配置有任何问题，请告诉我？请在下面找到我的 application.config 文件：

abc-management {
  kafka {
    capture-topic = [${?ABC}, "abc.pqr"]
    producer {
      brokers  =  ["abc123.systems.pqr:3230,abc12314.systems.pqr:3230,abc11234.systems.pqr:1244"]
    }
    consumer {
      zkConnect  = ["abc12443.systems.abc:234","abc1244.systems.abc:1241","abc121414.systems.abc:2181"]
      groupid = ["abc-consumer"]
      auto-offset-reset = "earliest"
    }
    offset {
      group-batch = 5
      group-time = 3
      parallelism-factor = 2
    }
  }
  akka {
    actor {
      cluster-name = "ABCSystem"
      timeout = [${?TIMEOUT}, 10]
      supervisor-strategy {
        max-number-of-retries = 10
        within-time-range = 30
      }
    }
  }
  shutdown-hook-time = 30
}
akka {
  actor {
    provider = "akka.cluster.ClusterActorRefProvider"
    deployment {
      /abcActor {
        router = round-robin-pool
        nr-of-instances = 20
      }
    }
  }
  kafka {
    consumer {
      poll-interval = 30ms
      poll-timeout = 30ms
      stop-timeout = 20s
      close-timeout = 10s
      commit-timeout = 10s
      wakeup-timeout = 10s
      use-dispatcher = "akka.kafka.default-dispatcher"
      kafka-clients {
        enable.auto.commit = true
        security.protocol = "SASL_PLAINTEXT"
         #sasl.kerberos.service.name=kafka
         #ssl.client.auth = "none"
      }
    }
    producer{
      kafka-clients {
         security.protocol = "SASL_PLAINTEXT"
         #sasl.kerberos.service.name=kafka
        # ssl.client.auth = "none"
      }
    }
  }
  remote {
    log-remote-lifecycle-events = off
    netty.tcp {
      hostname = "127.0.0.1"
      port = 2551
    }
  }
  cluster {
    seed-host = "127.0.0.1"
    seed-host = ${?AKKA_SEED_HOST}
    seed-port = "2551"
    seed-port = ${?AKKA_SEED_PORT}
    seed-nodes = [
      "akka.tcp://"${abc-management.akka.actor.cluster-name}"@"${akka.cluster.seed-host}":"${akka.cluster.seed-port}
    ]
    min-nr-of-members = 1
  }
}
# Disable legacy metrics in akka-cluster.
akka.cluster.metrics.enabled=off
# Enable metrics extension in akka-cluster-metrics.
akka.extensions=["akka.cluster.metrics.ClusterMetricsExtension"]
circuit-breaker {
  maxFailures = [${?CB_MAX_FAILURES}, 10]
  callTimeout = [${?CB_MAX_FAILURES}, 10000]
  resetTimeout = [${?CB_MAX_FAILURES}, 30000]
}

下面是我的 jaas 文件：

KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
doNotPrompt=true
useTicketCache=true
useKeyTab=true
keyTab="/home/abc/.abc.headless.keytab"
renewTicket=true
serviceName="kafka"
principal="abc@ABC.ADROOT.PQR"
debug=true
client=true;
};
Client {
  com.sun.security.auth.module.Krb5LoginModule required
  doNotPrompt=true
  useKeyTab=true
  storeKey=true
  useTicketCache=false
  serviceName="zookeeper"
  keyTab="/home/abc/.abc.headless.keytab"
  principal="abc@ABC.ADROOT.PQR"
  debug=true;
};

请在下面找到消费者端日志：

[xyz@asd12344 lib]$ java -Dconfig.file=application.conf -Djava.security.auth.login.config=kafka_jaas.conf -jar abc-management.jar
[INFO] [06/07/2017 16:29:12.354] [main] [akka.remote.Remoting] Starting remoting
[INFO] [06/07/2017 16:29:12.492] [main] [akka.remote.Remoting] Remoting started; listening on addresses :[akka.tcp://ABCSystem@127.0.0.1:2551]
[INFO] [06/07/2017 16:29:12.503] [main] [akka.cluster.Cluster(akka://ABCSystem)] Cluster Node [akka.tcp://ABCSystem@127.0.0.1:2551] - Starting up...
[INFO] [06/07/2017 16:29:12.586] [main] [akka.cluster.Cluster(akka://ABCSystem)] Cluster Node [akka.tcp://ABCSystem@127.0.0.1:2551] - Registered cluster JMX MBean [akka:type=Cluster]
[INFO] [06/07/2017 16:29:12.586] [main] [akka.cluster.Cluster(akka://ABCSystem)] Cluster Node [akka.tcp://ABCSystem@127.0.0.1:2551] - Started up successfully
[INFO] [06/07/2017 16:29:12.619] [ABCSystem-akka.actor.default-dispatcher-15] [akka.cluster.Cluster(akka://ABCSystem)] Cluster Node [akka.tcp://ABCSystem@127.0.0.1:2551] - Metrics collection has started successfully
[INFO] [06/07/2017 16:29:12.629] [ABCSystem-akka.actor.default-dispatcher-3] [akka.cluster.Cluster(akka://ABCSystem)] Cluster Node [akka.tcp://ABCSystem@127.0.0.1:2551] - Node [akka.tcp://ABCSystem@127.0.0.1:2551] is JOINING, roles []
[INFO] [06/07/2017 16:29:12.637] [ABCSystem-akka.actor.default-dispatcher-3] [akka.cluster.Cluster(akka://ABCSystem)] Cluster Node [akka.tcp://ABCSystem@127.0.0.1:2551] - Leader is moving node [akka.tcp://ABCSystem@127.0.0.1:2551] to [Up]
ERROR StatusLogger No log4j2 configuration file found. Using default configuration: logging only errors to the console.
[INFO] [06/07/2017 16:29:44.498] [ABCSystem-akka.actor.default-dispatcher-24] [akka://ABCSystem/user/abcManagementActor/$h/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.498] [ABCSystem-akka.actor.default-dispatcher-30] [akka://ABCSystem/user/abcManagementActor/$c/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.497] [ABCSystem-akka.actor.default-dispatcher-22] [akka://ABCSystem/user/abcManagementActor/$i/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.501] [ABCSystem-akka.actor.default-dispatcher-22] [akka://ABCSystem/user/abcManagementActor/$d/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.501] [ABCSystem-akka.actor.default-dispatcher-28] [akka://ABCSystem/user/abcManagementActor/$b/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.499] [ABCSystem-akka.actor.default-dispatcher-23] [akka://ABCSystem/user/abcManagementActor/$f/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.502] [ABCSystem-akka.actor.default-dispatcher-26] [akka://ABCSystem/user/abcManagementActor/$a/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.502] [ABCSystem-akka.actor.default-dispatcher-25] [akka://ABCSystem/user/abcManagementActor/$e/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.504] [ABCSystem-akka.actor.default-dispatcher-29] [akka://ABCSystem/user/abcManagementActor/$g/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.505] [ABCSystem-akka.actor.default-dispatcher-18] [akka://ABCSystem/user/abcManagementActor/$s/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.507] [ABCSystem-akka.actor.default-dispatcher-44] [akka://ABCSystem/user/abcManagementActor/$q/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.507] [ABCSystem-akka.actor.default-dispatcher-44] [akka://ABCSystem/user/abcManagementActor/$j/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.507] [ABCSystem-akka.actor.default-dispatcher-45] [akka://ABCSystem/user/abcManagementActor/$m/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.508] [ABCSystem-akka.actor.default-dispatcher-45] [akka://ABCSystem/user/abcManagementActor/$l/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.508] [ABCSystem-akka.actor.default-dispatcher-29] [akka://ABCSystem/user/abcManagementActor/$n/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.509] [ABCSystem-akka.actor.default-dispatcher-47] [akka://ABCSystem/user/abcManagementActor/$k/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.509] [ABCSystem-akka.actor.default-dispatcher-18] [akka.tcp://ABCSystem@127.0.0.1:2551/user/abcManagementActor/$f] Member up: akka.tcp://ABCSystem@127.0.0.1:2551
[INFO] [06/07/2017 16:29:44.510] [ABCSystem-akka.actor.default-dispatcher-56] [akka://ABCSystem/user/abcManagementActor/$p/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.510] [ABCSystem-akka.actor.default-dispatcher-2] [akka://ABCSystem/user/abcManagementActor/$r/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.510] [ABCSystem-akka.actor.default-dispatcher-21] [akka://ABCSystem/user/abcManagementActor/$o/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.512] [ABCSystem-akka.actor.default-dispatcher-15] [akka://ABCSystem/user/abcManagementActor/$t/abcEnrichmentActor] The abcEnrichment Actor is ready to receive the requests
[INFO] [06/07/2017 16:29:44.512] [ABCSystem-akka.actor.default-dispatcher-15] [akka.tcp://ABCSystem@127.0.0.1:2551/user/abcManagementActor/$a] Member up: akka.tcp://ABCSystem@127.0.0.1:2551
[INFO] [06/07/2017 16:29:44.513] [ABCSystem-akka.actor.default-dispatcher-4] [akka.tcp://ABCSystem@127.0.0.1:2551/user/abcManagementActor/$i] Member up: akka.tcp://ABCSystem@127.0.0.1:2551
[INFO] [06/07/2017 16:29:44.516] [ABCSystem-akka.actor.default-dispatcher-25] [akka.tcp://ABCSystem@127.0.0.1:2551/user/abcManagementActor/$g] Member up: akka.tcp://ABCSystem@127.0.0.1:2551
[INFO] [06/07/2017 16:29:44.517] [ABCSystem-akka.actor.default-dispatcher-4] [akka.tcp://ABCSystem@127.0.0.1:2551/user/abcManagementActor/$e] Member up: akka.tcp://ABCSystem@127.0.0.1:2551
Debug is  true storeKey false useTicketCache true useKeyTab true doNotPrompt true ticketCache is null isInitiator true KeyTab is /home/xyz/.xyz.headless.keytab refreshKrb5Config is false principal is xyz@ABC.ADROOT.PQR tryFirstPass is false useFirstPass is false storePass is false clearPass is false
Acquire TGT from Cache
Principal is xyz@ABC.ADROOT.PQR
Commit Succeeded

[WARN] [06/07/2017 16:30:04.830] [ABCSystem-akka.kafka.default-dispatcher-62] [akka.tcp://ABCSystem@127.0.0.1:2551/system/kafka-consumer-1] Consumer interrupted with WakeupException after timeout. Message: null. Current value of akka.kafka.consumer.wakeup-timeout is 20000 milliseconds
[WARN] [06/07/2017 16:30:24.919] [ABCSystem-akka.kafka.default-dispatcher-64] [akka.tcp://ABCSystem@127.0.0.1:2551/system/kafka-consumer-1] Consumer interrupted with WakeupException after timeout. Message: null. Current value of akka.kafka.consumer.wakeup-timeout is 20000 milliseconds
^C16:30:35.253 [shutdownHook1] INFO  c.pqr.asd.abcManagementMain - Terminating... - 2017-06-07T15:30:35.239Z
[INFO] [06/07/2017 16:30:35.265] [ABCSystem-akka.actor.default-dispatcher-27] [akka://ABCSystem/user/abcManagementActor/$b/abcEnrichmentActor] The abcEnrichment Actor is gonna stop and would not entertain any requests
[INFO] [06/07/2017 16:30:35.265] [ABCSystem-akka.actor.default-dispatcher-16] [akka://ABCSystem/user/abcManagementActor/$a/abcEnrichmentActor] The abcEnrichment Actor is gonna stop and would not entertain any requests
o.a.k.c.s.kerberos.KerberosLogin - [Principal=xyz@ABC.ADROOT.pqr]: TGT renewal thread has been interrupted and will exit.
16:30:45.020 [shutdownHook1] INFO  c.pqr.asd.abcManagementMain - Terminated... Bye - 2017-06-07T15:30:45.020Z
[xyz@jbt13993 lib]$

【问题讨论】：

好的，所以你在消费者获得 Kerberos TGT（通用身份验证阶段）后不到 1 秒内强制向消费者发送 KILL 信号 (Ctrl-C)；并且您没有启用完整的 Kerberos 调试跟踪，该跟踪会显示 Kerberos 服务票证发生的情况。这证明了什么？
顺便问一下，你有什么问题吗？然后明确...
@SamsonScharfrichter：我有疑问，这就是我在这里发布它的原因；我在 Kafka Broker 上从生产者端发送消息，这些消息没有在 Kafka 消费者端被消耗，这背后的原因我无法根据你的建议找到我将尝试启用完整的 kerberos 调试跟踪
这仍然不是一个问题，它是一个断言。
这里我无法从消费者端的 scala akka kafka 生产者发送消息......所以我认为我发送数据或配置的方式可能存在一些错误我无法找出错误所以我问有人能帮我找出我做错了什么吗？

标签： scala security akka apache-kafka kerberos

【解决方案1】：

正如我所说，我们无法在开发集群的消费者端接收消息，我们能够在本地机器上接收消息，这是因为如果您仔细检查，我们的开发集群 Kafka 版本是 0.9，而我的本地机器 kafka 版本是 0.10 Kafka 0.9 和 kafka 0.10 版本之间存在显着差异。更改我们的 API 和版本后，它可以正常工作。

【讨论】：