ksqlDB is not working at all

[sbourell]

Hello everyone,

I was wondering if anyone have a working AKHQ with a Confluent kSqlDB connected on it that is working ?

Because I try all ways I could find on the web, this github and all the issues that talks about ksql and it still not working for me.

For reference, this is my configuration:

akhq:
  connections:
    dev:
      properties:
        bootstrap.servers: 'pkc-XXXXX.eu-central-1.aws.confluent.cloud:9092'
        security.protocol: SASL_SSL
        ssl.endpoint.identification.algorithm: https
        sasl.mechanism: PLAIN
        sasl.jaas.config: >-
          org.apache.kafka.common.security.plain.PlainLoginModule required
          username="XXXXXXXXXXXXXX"
          password="XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX";
      schema-registry:
        url: https://psrc-XXXXXX.eu-central-1.aws.confluent.cloud:8085
        type: "confluent"
        basic-auth-username: "XXXXXXXXXXXXXX"
        basic-auth-password: "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
      ksqldb:
        - name: "ksql-dev"
          url: https://pksqlc-XXXXXX.eu-central-1.aws.confluent.cloud:443
          useTLS: true
          basic-auth-username: "XXXXXXXXXXXXXX"
          basic-auth-password: "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX"

And anytime I try to access it, I have this error:

2024-09-24 08:07:50,286 �[1;31mERROR�[0;39m �[35mvert.x-eventloop-thread-1�[0;39m �[36mio.vertx.core.impl.ContextBase�[0;39m Unhandled exception
java.lang.NullPointerException: Cannot invoke "io.vertx.core.http.HttpClientRequest.response(io.vertx.core.Handler)" because "request" is null
	at io.confluent.ksql.api.client.impl.ClientImpl.lambda$makeRequest$29(ClientImpl.java:746)
	at io.vertx.core.impl.future.FutureImpl$3.onFailure(FutureImpl.java:153)
	at io.vertx.core.impl.future.FutureBase.emitFailure(FutureBase.java:75)
	at io.vertx.core.impl.future.FutureImpl.tryFail(FutureImpl.java:230)
	at io.vertx.core.impl.future.PromiseImpl.tryFail(PromiseImpl.java:23)
	at io.vertx.core.http.impl.HttpClientImpl.lambda$doRequest$6(HttpClientImpl.java:689)
	at io.vertx.core.net.impl.pool.Endpoint.lambda$getConnection$0(Endpoint.java:52)
	at io.vertx.core.http.impl.SharedClientHttpStreamEndpoint$Request.handle(SharedClientHttpStreamEndpoint.java:162)
	at io.vertx.core.http.impl.SharedClientHttpStreamEndpoint$Request.handle(SharedClientHttpStreamEndpoint.java:123)
	at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:55)
	at io.vertx.core.impl.ContextBase.emit(ContextBase.java:297)
	at io.vertx.core.net.impl.pool.SimpleConnectionPool$ConnectFailed$1.run(SimpleConnectionPool.java:381)
	at io.vertx.core.net.impl.pool.Task.runNextTasks(Task.java:43)
	at io.vertx.core.net.impl.pool.CombinerExecutor.submit(CombinerExecutor.java:82)
	at io.vertx.core.net.impl.pool.SimpleConnectionPool.execute(SimpleConnectionPool.java:245)
	at io.vertx.core.net.impl.pool.SimpleConnectionPool.lambda$connect$2(SimpleConnectionPool.java:259)
	at io.vertx.core.http.impl.SharedClientHttpStreamEndpoint.lambda$connect$2(SharedClientHttpStreamEndpoint.java:104)
	at io.vertx.core.impl.future.FutureImpl$3.onFailure(FutureImpl.java:153)
	at io.vertx.core.impl.future.FutureBase.emitFailure(FutureBase.java:75)
	at io.vertx.core.impl.future.FutureImpl.tryFail(FutureImpl.java:230)
	at io.vertx.core.impl.future.Composition$1.onFailure(Composition.java:66)
	at io.vertx.core.impl.future.FutureBase.lambda$emitFailure$1(FutureBase.java:69)
	at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:86)
	at io.vertx.core.impl.ContextBase.execute(ContextBase.java:283)
	at io.vertx.core.impl.future.FutureBase.emitFailure(FutureBase.java:66)
	at io.vertx.core.impl.future.FutureImpl.tryFail(FutureImpl.java:230)
	at io.vertx.core.impl.future.PromiseImpl.tryFail(PromiseImpl.java:23)
	at io.vertx.core.Promise.fail(Promise.java:89)
	at io.vertx.core.impl.future.PromiseImpl.operationComplete(PromiseImpl.java:67)
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:590)
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:557)
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:492)
	at io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:636)
	at io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:629)
	at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:118)
	at io.vertx.core.http.impl.VertxHttp2ConnectionHandler.channelInactive(VertxHttp2ConnectionHandler.java:147)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:303)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
	at io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:274)
	at io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:301)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:281)
	at io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901)
	at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Unknown Source)

I try to look at the source code with my knowledge (I am not a Java developer) and I don't understand this:
https://github.com/tchiotludo/akhq/blob/0.25.1/src/main/java/org/akhq/modules/KafkaModule.java

I cannot find any reference for this method in the class definition nor in the confluent packages.

Is someone able to tell me how this is supposed to work ?

Thank you for your help !
Stephane

[Zico56]

Hello,

This is not working because for Confluent Cloud case, some flags are required for TLS connection.
Setting "useTls: true" and "useAlpn: true" in your application yml file will fix the issue.

Regards.

[sbourell]

Thank you @Zico56 it's working now!
Glad you stopped by 😉

Maybe somebody can update the documentation or the example configuration file ?

Working configuration for reference:

akhq:
  connections:
    {NAME}:
      properties:
        bootstrap.servers: '{SERVER_URL}:9092'
        security.protocol: SASL_SSL
        ssl.endpoint.identification.algorithm: https
        sasl.mechanism: PLAIN
        sasl.jaas.config: >-
          org.apache.kafka.common.security.plain.PlainLoginModule required
          username="{ACCESS_KEY}"
          password="{SECRET_KEY}";
      ksqldb:
        - name: "{NAME}"
          url: https://{CONFLUENT_URI}.confluent.cloud:443
          basic-auth-username: "{ACCESS_KEY}"
          basic-auth-password: "{SECRET_KEY}"
          useTls: true
          useAlpn: true