kafka-rest
kafka-rest copied to clipboard
confluentinc
KAFKA REST SASL issue for consumer
If in our server.properties, we have not configured any plaintext port, we get the following error while using rest consumer. We are using only SASL_PLAINTEXT port.
[2018-03-07 15:06:53,709] WARN [test-consume-group_mdfdevapp3-1520415401800-2daebe03-leader-finder-thread]: Failed to find leader for Set(topic-1-0) (kafka.consumer.ConsumerFetcherManager$LeaderFinderThread:93) kafka.common.BrokerEndPointNotAvailableException: End point with security protocol PLAINTEXT not found for broker 0 at kafka.client.ClientUtils$$anonfun$getPlaintextBrokerEndPoints$1$$anonfun$apply$5.apply(ClientUtils.scala:149) at kafka.client.ClientUtils$$anonfun$getPlaintextBrokerEndPoints$1$$anonfun$apply$5.apply(ClientUtils.scala:149) at scala.Option.getOrElse(Option.scala:121) at kafka.client.ClientUtils$$anonfun$getPlaintextBrokerEndPoints$1.apply(ClientUtils.scala:149) at kafka.client.ClientUtils$$anonfun$getPlaintextBrokerEndPoints$1.apply(ClientUtils.scala:145) at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at scala.collection.TraversableLike$class.map(TraversableLike.scala:234) at scala.collection.AbstractTraversable.map(Traversable.scala:104) at kafka.client.ClientUtils$.getPlaintextBrokerEndPoints(ClientUtils.scala:145) at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:64)
If we have a plaintext port configured in server.properties, I am getting the following error while using rest consume messages.
[2018-03-07 15:11:14,418] INFO Connected to 135.250.193.237:9093 for producing (kafka.producer.SyncProducer:72) [2018-03-07 15:11:14,419] INFO Disconnecting from 135.250.193.237:9093 (kafka.producer.SyncProducer:72) [2018-03-07 15:11:14,419] INFO [ConsumerFetcherManager-1520415663046] Added fetcher for partitions ArrayBuffer() (kafka.consumer.ConsumerFetcherManager:72) [2018-03-07 15:11:14,625] INFO Verifying properties (kafka.utils.VerifiableProperties:72) [2018-03-07 15:11:14,625] INFO Property client.id is overridden to test-consume-group (kafka.utils.VerifiableProperties:72) [2018-03-07 15:11:14,625] INFO Property metadata.broker.list is overridden to 135.250.193.237:9093 (kafka.utils.VerifiableProperties:72) [2018-03-07 15:11:14,625] INFO Property request.timeout.ms is overridden to 30000 (kafka.utils.VerifiableProperties:72) [2018-03-07 15:11:14,625] INFO Fetching metadata from broker BrokerEndPoint(0,135.250.193.237,9093) with correlation id 26 for 1 topic(s) Set(topic-1) (kafka.client.ClientUtils$:72)
Can you let us know why we cannot consume using SASL port for REST queries ? Is there any reason why the port is looked up in server.properties file ?
Thanks!
Does consumer group ACL will be checked when using kafka rest to consumer messages? I configure kafka-rest with: zookeeper.connect=localhost:2181 bootstrap.servers=SASL_PLAINTEXT://localhost:9093 client.security.protocol=SASL_PLAINTEXT client.sasl.mechanism=PLAIN But the consumer group ACL doesn't be checked. Since the message in kafka broker is read as ANONYMOUS. This is used by PLAINTEXT, but I've configured with SASL_PLAINTEXT.
I found that Kafka REST Security plugin and ACLs does not work with the V1 consumer API’s since it uses a simple consumer which doesn’t support Kafka Security. And in V2, no API is used to consume messages.
I'd like to know any deliver plan for the ACL of consumer by Kafka Rest.
