无法使用akka-kafka kerberos配置在kafka使用者端接收kafka消息

ercv8c1e  于 2021-06-08  发布在  Kafka
关注(0)|答案(1)|浏览(333)

我正在用akka、kafka、scala开发应用程序，我的应用程序在本地系统上运行良好，但每当我在启用kerberos安全的集群上部署它时，我就无法在kafka使用者端接收消息。为了测试的目的，我从kafka命令行运行producer并从命令行发送消息。如果我的配置有任何错误，请告诉我好吗？请在下面找到my 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]$
scalaakkaapache-kafkakerberossecurity

1条答案

按热度按时间
cclgggtu

cclgggtu1#

正如我所说，我们无法在dev cluster上的consumer端接收消息，我们能够在本地计算机上接收消息，这是因为我们的dev cluster kafka版本是0.9，而我的本地计算机kafka版本是0.10。如果您仔细检查，kafka 0.9和kafka 0.10版本之间存在显著差异。更改我们的api的版本和工作正常。

赞(0）分享  回复(0）举报  2021-06-09
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com