NiFi - Unable to connect to Kafka with different/weak encryption

Issue

KafkaPublish and ConsumerKafka Processor in NiFi with strong encryption (aes128/aes256) cannot connect to Kafka Server with weak encryption (rc4-hmac) with following error:

2023-11-23 14:00:14,523 INFO org.apache.kafka.clients.producer.ProducerConfig: ProducerConfig values:
        acks = -1
        batch.size = 16384
        bootstrap.servers = [node49.example.com:9093, node50.example.com:9093, node51.example.com:9093, node52.example.com:9093, node53.example.com:9093, node54.example.com:9093, node115.example.com:9093, node116.example.com:9093, node117.example.com:9093, node118.example.com:9093, node119.example.com:9093, node120.example.com:9093]
        buffer.memory = 33554432
        client.dns.lookup = use_all_dns_ips
        client.id = producer-7
        compression.type = none
        connections.max.idle.ms = 540000
        delivery.timeout.ms = 302000
        enable.idempotence = false
        interceptor.classes = []
        internal.auto.downgrade.txn.commit = false
        key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
        linger.ms = 500
        max.block.ms = 5000
        max.in.flight.requests.per.connection = 5
        max.request.size = 1048576
        metadata.max.age.ms = 300000
        metadata.max.idle.ms = 300000
        metric.reporters = []
        metrics.num.samples = 2
        metrics.recording.level = INFO
        metrics.sample.window.ms = 30000
        partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
        receive.buffer.bytes = 32768
        reconnect.backoff.max.ms = 1000
        reconnect.backoff.ms = 50
        request.timeout.ms = 60000
        retries = 5
        retry.backoff.ms = 100
        sasl.client.callback.handler.class = null
        sasl.jaas.config = [hidden]
        sasl.kerberos.kinit.cmd = /usr/bin/kinit
        sasl.kerberos.min.time.before.relogin = 60000
        sasl.kerberos.service.name = kafka
        sasl.kerberos.ticket.renew.jitter = 0.05
        sasl.kerberos.ticket.renew.window.factor = 0.8
        sasl.login.callback.handler.class = null
        sasl.login.class = class org.apache.nifi.processors.kafka.pubsub.CustomKerberosLogin
        sasl.login.refresh.buffer.seconds = 300
        sasl.login.refresh.min.period.seconds = 60
        sasl.login.refresh.window.factor = 0.8
        sasl.login.refresh.window.jitter = 0.05
        sasl.mechanism = GSSAPI
        security.protocol = SASL_SSL
        security.providers = null
        send.buffer.bytes = 131072
        ssl.cipher.suites = null
        ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
        ssl.endpoint.identification.algorithm = https
        ssl.engine.factory.class = null
        ssl.key.password = null
        ssl.keymanager.algorithm = SunX509
        ssl.keystore.location = null
        ssl.keystore.password = null
        ssl.keystore.type = JKS
        ssl.protocol = TLSv1.3
        ssl.provider = null
        ssl.secure.random.implementation = null
        ssl.trustmanager.algorithm = PKIX
        ssl.truststore.location = /opt/cloudera/security/jks/truststore.jks
        ssl.truststore.password = [hidden]
        ssl.truststore.type = JKS
        transaction.timeout.ms = 900000
        transactional.id = null
        value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

2023-11-23 14:00:14,524 INFO org.apache.kafka.clients.producer.KafkaProducer: [Producer clientId=producer-7] Closing the Kafka producer with timeoutMillis = 0 ms.
2023-11-23 14:00:14,524 ERROR org.apache.nifi.processors.kafka.pubsub.PublishKafkaRecord_2_6: PublishKafkaRecord_2_6[id=07b9361d-337c-1833-98e0-a1db3cc3fffb] Processing halted: yielding [1 sec]
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:441)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:273)
        at org.apache.nifi.processors.kafka.pubsub.PublisherPool.createLease(PublisherPool.java:88)
        at org.apache.nifi.processors.kafka.pubsub.PublisherPool.obtainPublisher(PublisherPool.java:78)
        at org.apache.nifi.processors.kafka.pubsub.PublishKafkaRecord_2_6.onTrigger(PublishKafkaRecord_2_6.java:514)
        at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)
        at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1361)
        at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:247)
        at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:102)
        at org.apache.nifi.engine.FlowEngine$2.run(FlowEngine.java:110)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
        at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: Could not login: the client is being asked for a password, but the Kafka client code does not currently support obtaining a password from the user. not available to garner  authentication information from the user
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:172)
        at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:157)
        at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:73)
        at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105)
        at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:449)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:430)
        ... 15 common frames omitted
Caused by: javax.security.auth.login.LoginException: Could not login: the client is being asked for a password, but the Kafka client code does not currently support obtaining a password from the user. not available to garner  authentication information from the user
        at jdk.security.auth/com.sun.security.auth.module.Krb5LoginModule.promptForPass(Krb5LoginModule.java:925)
        at jdk.security.auth/com.sun.security.auth.module.Krb5LoginModule.attemptAuthentication(Krb5LoginModule.java:745)
        at jdk.security.auth/com.sun.security.auth.module.Krb5LoginModule.login(Krb5LoginModule.java:597)
        at java.base/javax.security.auth.login.LoginContext.invoke(LoginContext.java:755)
        at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:679)
        at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:677)
        at java.base/java.security.AccessController.doPrivileged(AccessController.java:712)
        at java.base/javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:677)
        at java.base/javax.security.auth.login.LoginContext.login(LoginContext.java:587)
        at org.apache.kafka.common.security.authenticator.AbstractLogin.login(AbstractLogin.java:60)
        at org.apache.nifi.processors.kafka.pubsub.CustomKerberosLogin.login(CustomKerberosLogin.java:82)
        at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:62)
        at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:105)
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:158)
        ... 20 common frames omitted
2023-11-23 14:00:14,524 WARN org.apache.nifi.controller.tasks.ConnectableTask: Processing halted: uncaught exception in Component [PublishKafkaRecord_2_6[id=07b9361d-337c-1833-98e0-a1db3cc3fffb]]
org.apache.kafka.common.KafkaException: Failed to construct kafka producer
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:441)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:273)
        at org.apache.nifi.processors.kafka.pubsub.PublisherPool.createLease(PublisherPool.java:88)
        at org.apache.nifi.processors.kafka.pubsub.PublisherPool.obtainPublisher(PublisherPool.java:78)
        at org.apache.nifi.processors.kafka.pubsub.PublishKafkaRecord_2_6.onTrigger(PublishKafkaRecord_2_6.java:514)
        at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)
        at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1361)
        at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:247)
        at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:102)
        at org.apache.nifi.engine.FlowEngine$2.run(FlowEngine.java:110)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
        at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.common.KafkaException: javax.security.auth.login.LoginException: Could not login: the client is being asked for a password, but the Kafka client code does not currently support obtaining a password from the user. not available to garner  authentication information from the user
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:172)
        at org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:157)
        at org.apache.kafka.common.network.ChannelBuilders.clientChannelBuilder(ChannelBuilders.java:73)
        at org.apache.kafka.clients.ClientUtils.createChannelBuilder(ClientUtils.java:105)
        at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:449)
        at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:430)
        ... 15 common frames omitted
Caused by: javax.security.auth.login.LoginException: Could not login: the client is being asked for a password, but the Kafka client code does not currently support obtaining a password from the user. not available to garner  authentication information from the user
        at jdk.security.auth/com.sun.security.auth.module.Krb5LoginModule.promptForPass(Krb5LoginModule.java:925)
        at jdk.security.auth/com.sun.security.auth.module.Krb5LoginModule.attemptAuthentication(Krb5LoginModule.java:745)
        at jdk.security.auth/com.sun.security.auth.module.Krb5LoginModule.login(Krb5LoginModule.java:597)
        at java.base/javax.security.auth.login.LoginContext.invoke(LoginContext.java:755)
        at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:679)
        at java.base/javax.security.auth.login.LoginContext$4.run(LoginContext.java:677)
        at java.base/java.security.AccessController.doPrivileged(AccessController.java:712)
        at java.base/javax.security.auth.login.LoginContext.invokePriv(LoginContext.java:677)
        at java.base/javax.security.auth.login.LoginContext.login(LoginContext.java:587)
        at org.apache.kafka.common.security.authenticator.AbstractLogin.login(AbstractLogin.java:60)
        at org.apache.nifi.processors.kafka.pubsub.CustomKerberosLogin.login(CustomKerberosLogin.java:82)
        at org.apache.kafka.common.security.authenticator.LoginManager.<init>(LoginManager.java:62)
        at org.apache.kafka.common.security.authenticator.LoginManager.acquireLoginManager(LoginManager.java:105)
        at org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:158)
        ... 20 common frames omitted

Cause

  • NiFi is configured with aes encryption only. While kafka server is still using rc4 encryption.

Resolution

  • Add "rc4-hmac" in CM > Administration > Kerberos > krb_enc_types

  • Regenerate credential if keytab still not have rc4-hmac encryption

      # klist -ekt /var/lib/nifi/nifi.keytab
    

    Make sure "rc4-hmac" entries is exist.

  • Allow "rc4-hmac" in /etc/krb5.conf

  • Add "allow_weak_crypto = true" in /etc/krb5.conf