Statefun + Confluent Fully-managed Kafka

Posted by hezekiah maina on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Statefun-Confluent-Fully-managed-Kafka-tp38532.html

Hi,

I'm trying to use Stateful Functions with Kafka as my ingress and egress. I'm using the Confluent fully-managed Kafka and I'm having a challenge adding my authentication details in the module.yaml file.
Here is my current config details:
version: "1.0"
module:
  meta:
    type: remote
  spec:
    functions:
      - function:
          meta:
            kind: http
            type: example/greeter
          spec:
            endpoint: <https-endpoint>
            states:
              - seen_count
            maxNumBatchRequests: 500
            timeout: 2min
    ingresses:
      - ingress:
          meta:
            type: statefun.kafka.io/routable-protobuf-ingress
            id: example/names
          spec:
            address: <confluent-bootstrap-server>
            consumerGroupId: statefun-consumer-group
            topics:
              - topic: names
                typeUrl: com.googleapis/example.GreetRequest
                targets:
                  - example/greeter
            properties:
              - bootstrap.servers:<confluent-bootstrap-server>
              - security.protocol: SASL_SSL
              - sasl.mechanism: PLAIN
              - sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";
              - ssl.endpoint.identification.algorithm: https
    egresses:
      - egress:
          meta:
            type: statefun.kafka.io/generic-egress
            id: example/greets
          spec:
            address: <confluent-bootstrap-server>
            deliverySemantic:
              type: exactly-once
              transactionTimeoutMillis: 100000
            properties:
              - bootstrap.servers: <confluent-bootstrap-server>
              - security.protocol: SASL_SSL
              - sasl.mechanisms: PLAIN
              - sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="USERNAME" password="PASSWORD";
              - ssl.endpoint.identification.algorithm: https

After running docker-compose with a master and worker containers I'm getting this error:

Could not find a 'KafkaClient' entry in the JAAS configuration. System property 'java.security.auth.login.config' is /tmp/jaas-2846080966990890307.conf

The producer config logged :
worker_1  | 2020-10-07 13:38:08,489 INFO  org.apache.kafka.clients.producer.ProducerConfig              - ProducerConfig values:
worker_1  |     acks = 1
worker_1  |     batch.size = 16384
worker_1  |     bootstrap.servers = [https://---.asia-southeast1.gcp.confluent.cloud:9092]
worker_1  |     buffer.memory = 33554432
worker_1  |     client.dns.lookup = default
worker_1  |     client.id =
worker_1  |     compression.type = none
worker_1  |     connections.max.idle.ms = 540000
worker_1  |     delivery.timeout.ms = 120000
worker_1  |     enable.idempotence = false
worker_1  |     interceptor.classes = []
worker_1  |     key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
worker_1  |     linger.ms = 0
worker_1  |     max.block.ms = 60000
worker_1  |     max.in.flight.requests.per.connection = 5
worker_1  |     max.request.size = 1048576
worker_1  |     metadata.max.age.ms = 300000
worker_1  |     metric.reporters = []
worker_1  |     metrics.num.samples = 2
worker_1  |     metrics.recording.level = INFO
worker_1  |     metrics.sample.window.ms = 30000
worker_1  |     partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
worker_1  |     receive.buffer.bytes = 32768
worker_1  |     reconnect.backoff.max.ms = 1000
worker_1  |     reconnect.backoff.ms = 50
worker_1  |     request.timeout.ms = 30000
worker_1  |     retries = 2147483647
worker_1  |     retry.backoff.ms = 100
worker_1  |     sasl.client.callback.handler.class = null
worker_1  |     sasl.jaas.config = null
worker_1  |     sasl.kerberos.kinit.cmd = /usr/bin/kinit
worker_1  |     sasl.kerberos.min.time.before.relogin = 60000
worker_1  |     sasl.kerberos.service.name = null
worker_1  |     sasl.kerberos.ticket.renew.jitter = 0.05
worker_1  |     sasl.kerberos.ticket.renew.window.factor = 0.8
worker_1  |     sasl.login.callback.handler.class = null
worker_1  |     sasl.login.class = null
worker_1  |     sasl.login.refresh.buffer.seconds = 300
worker_1  |     sasl.login.refresh.min.period.seconds = 60
worker_1  |     sasl.login.refresh.window.factor = 0.8
worker_1  |     sasl.login.refresh.window.jitter = 0.05
worker_1  |     sasl.mechanism = GSSAPI
worker_1  |     security.protocol = SASL_SSL
worker_1  |     send.buffer.bytes = 131072
worker_1  |     ssl.cipher.suites = null
worker_1  |     ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
worker_1  |     ssl.endpoint.identification.algorithm = https
worker_1  |     ssl.key.password = null
worker_1  |     ssl.keymanager.algorithm = SunX509
worker_1  |     ssl.keystore.location = null
worker_1  |     ssl.keystore.password = null
worker_1  |     ssl.keystore.type = JKS
worker_1  |     ssl.protocol = TLS
worker_1  |     ssl.provider = null
worker_1  |     ssl.secure.random.implementation = null
worker_1  |     ssl.trustmanager.algorithm = PKIX
worker_1  |     ssl.truststore.location = null
worker_1  |     ssl.truststore.password = null
worker_1  |     ssl.truststore.type = JKS
worker_1  |     transaction.timeout.ms = 100000
worker_1  |     transactional.id = null
worker_1  |     value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer
worker_1  |

Is there something that I'm missing?