[
https://issues.apache.org/jira/browse/FLUME-2792?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15290734#comment-15290734
]
sankaranarayanan commented on FLUME-2792:
-----------------------------------------
We followed the same. but no luck. in our environment we enabled Ranger
application. the below error occurring.
16/05/19 05:18:47 INFO producer.SyncProducer: Connected to
[test1.example.com:6667:test2.example.com:6667:test3.example.com]:6667 for
producing
16/05/19 05:18:47 INFO producer.SyncProducer: Disconnecting from
[test1.example.com:6667:test2.example.com:6667:test3.example.com]:6667
16/05/19 05:18:47 WARN client.ClientUtils$: Fetching topic metadata with
correlation id 0 for topics [Set(testtopic1)] from broker
[id:0,host:test1.example.com:6667:test2.example.com:6667:test3.example.com,port:6667]
failed
java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:97)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:74)
at
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:73)
at kafka.producer.SyncProducer.send(SyncProducer.scala:114)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
at
kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
at
kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
at kafka.utils.Utils$.swallow(Utils.scala:172)
at kafka.utils.Logging$class.swallowError(Logging.scala:106)
at kafka.utils.Utils$.swallowError(Utils.scala:45)
at
kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
at kafka.producer.Producer.send(Producer.scala:76)
at kafka.javaapi.producer.Producer.send(Producer.scala:42)
at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:125)
at
org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
at java.lang.Thread.run(Thread.java:745)
16/05/19 05:18:47 INFO producer.SyncProducer: Disconnecting from
[test1.example.com:6667:test2.example.com:6667:test3.example.com]:6667
16/05/19 05:18:47 ERROR utils.Utils$: fetching topic metadata for topics
[Set(testtopic1)] from broker
[ArrayBuffer(id:0,host:test1.example.com:6667:test2.example.com:6667:test3.example.com,port:6667)]
failed
kafka.common.KafkaException: fetching topic metadata for topics
[Set(testtopic1)] from broker
[ArrayBuffer(id:0,host:test1.example.com:6667:test2.example.com:6667:test3.example.com,port:6667)]
failed
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:72)
at
kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
at
kafka.producer.async.DefaultEventHandler$$anonfun$handle$1.apply$mcV$sp(DefaultEventHandler.scala:67)
at kafka.utils.Utils$.swallow(Utils.scala:172)
at kafka.utils.Logging$class.swallowError(Logging.scala:106)
at kafka.utils.Utils$.swallowError(Utils.scala:45)
at
kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:67)
at kafka.producer.Producer.send(Producer.scala:76)
at kafka.javaapi.producer.Producer.send(Producer.scala:42)
at org.apache.flume.sink.kafka.KafkaSink.process(KafkaSink.java:125)
at
org.apache.flume.sink.DefaultSinkProcessor.process(DefaultSinkProcessor.java:68)
at org.apache.flume.SinkRunner$PollingRunner.run(SinkRunner.java:147)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.nio.channels.ClosedChannelException
at kafka.network.BlockingChannel.send(BlockingChannel.scala:97)
at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:74)
at
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:73)
at kafka.producer.SyncProducer.send(SyncProducer.scala:114)
at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
... 12 more
16/05/19 05:18:47 INFO client.ClientUtils$: Fetching metadata from broker
id:0,host:test1.example.com:6667:test2.example.com:6667:test3.example.com,port:6667
with correlation id 1 for 1 topic(s) Set(testtopic1)
Please help us.
> Flume Kafka Kerberos Support
> ----------------------------
>
> Key: FLUME-2792
> URL: https://issues.apache.org/jira/browse/FLUME-2792
> Project: Flume
> Issue Type: Bug
> Components: Configuration, Docs, Sinks+Sources
> Affects Versions: v1.6.0, v1.5.2
> Environment: HDP 2.3 fully kerberized including Kafka 0.8.2.2 + Flume
> 1.5.2 or Apache Flume 1.6 downloaded from apache.org
> Reporter: Hari Sekhon
> Priority: Blocker
>
> Following on from FLUME-2790 it appears as though Flume doesn't yet have
> support for Kafka + Kerberos as there are is no setting documented in the
> Flume 1.6.0 user guide under the Kafka source section to tell Flume to use
> plaintextsasl as the connection mechanism to Kafka and Kafka rejects
> unauthenticated plaintext mechanism:
> {code}15/09/10 16:51:22 INFO consumer.ConsumerFetcherManager:
> [ConsumerFetcherManager-1441903874830] Added fetcher for partitions
> ArrayBuffer()
> 15/09/10 16:51:22 WARN consumer.ConsumerFetcherManager$LeaderFinderThread:
> [flume_<custom_scrubbed>-1441903874763-abdc98ec-leader-finder-thread], Failed
> to find leader for Set([<custom_scrubbed>,0], [<custom_scrubbed>,1])
> kafka.common.BrokerEndPointNotAvailableException: End point PLAINTEXT not
> found for broker 0
> at kafka.cluster.Broker.getBrokerEndPoint(Broker.scala:140)
> at
> kafka.utils.ZkUtils$$anonfun$getAllBrokerEndPointsForChannel$1.apply(ZkUtils.scala:124)
> at
> kafka.utils.ZkUtils$$anonfun$getAllBrokerEndPointsForChannel$1.apply(ZkUtils.scala:124)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
> at
> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
> at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
> at scala.collection.AbstractTraversable.map(Traversable.scala:105)
> at
> kafka.utils.ZkUtils$.getAllBrokerEndPointsForChannel(ZkUtils.scala:124)
> at
> kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
> at
> kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:60){code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
