java.nio.BufferUnderflowException when displaying topic data

[BernhardBerbuir]

I am using the dev version of AKHQ in order to use the feature "fine granular permissions" with OIDC. This bug does not occur with version 0.24 and default-group: reader.

When I try to view the data of a topic, I see the following behaviour for some of the topics:

• the UI tries for ever to fetch data (but does not display an error message)
• the following exception occurs at the log of AKHQ:

java.lang.RuntimeException: Error for Consume with options {}
at org.akhq.utils.Debug.call(Debug.java:74)
at org.akhq.repositories.RecordRepository.consume(RecordRepository.java:133)
at org.akhq.controllers.TopicController.data(TopicController.java:230)
at org.akhq.controllers.$TopicController$Definition$Exec.dispatch(Unknown Source)
at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invoke(AbstractExecutableMethodsDefinition.java:371)
at io.micronaut.context.DefaultBeanContext$4.invoke(DefaultBeanContext.java:594)
at io.micronaut.web.router.AbstractRouteMatch.execute(AbstractRouteMatch.java:303)
at io.micronaut.web.router.RouteMatch.execute(RouteMatch.java:111)
at io.micronaut.http.context.ServerRequestContext.with(ServerRequestContext.java:103)
at io.micronaut.http.server.RouteExecutor.lambda$executeRoute$14(RouteExecutor.java:659)
at reactor.core.publisher.FluxDeferContextual.subscribe(FluxDeferContextual.java:49)
at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62)
at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194)
at io.micronaut.reactive.reactor.instrument.ReactorInstrumentation.lambda$init$0(ReactorInstrumentation.java:62)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
at io.micrometer.core.instrument.composite.CompositeTimer.recordCallable(CompositeTimer.java:129)
at io.micrometer.core.instrument.Timer.lambda$wrap$1(Timer.java:206)
at io.micronaut.scheduling.instrument.InvocationInstrumenterWrappedCallable.call(InvocationInstrumenterWrappedCallable.java:53)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.nio.BufferUnderflowException: null
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at java.base/jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(Unknown Source)
at java.base/jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source)
at java.base/java.lang.reflect.Constructor.newInstance(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.getThrowableException(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.reportException(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.invoke(Unknown Source)
at java.base/java.util.stream.Nodes.collect(Unknown Source)
at java.base/java.util.stream.ReferencePipeline.evaluateToNode(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source)
at java.base/java.util.stream.SortedOps$OfRef.opEvaluateParallel(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.opEvaluateParallelLazy(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.sourceSpliterator(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.evaluate(Unknown Source)
at java.base/java.util.stream.ReferencePipeline.collect(Unknown Source)
at org.akhq.repositories.RecordRepository.consumeNewest(RecordRepository.java:349)
at org.akhq.repositories.RecordRepository.lambda$consume$4(RecordRepository.java:139)
at org.akhq.utils.Debug.call(Debug.java:67)
... 22 common frames omitted
Caused by: java.nio.BufferUnderflowException: null
at java.base/java.nio.Buffer.nextGetIndex(Unknown Source)
at java.base/java.nio.HeapByteBuffer.getShort(Unknown Source)
at org.akhq.utils.ContentUtils.asShort(ContentUtils.java:64)
at org.akhq.utils.ContentUtils.convertToObject(ContentUtils.java:86)
at org.akhq.models.Record.<init>(Record.java:124)
at org.akhq.repositories.RecordRepository.newRecord(RecordRepository.java:491)
at org.akhq.repositories.RecordRepository.lambda$consumeNewest$14(RecordRepository.java:326)
at java.base/java.util.stream.ReferencePipeline$7$1.accept(Unknown Source)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)
at java.base/java.util.stream.ReferencePipeline$2$1.accept(Unknown Source)
at java.base/java.util.stream.ReferencePipeline$3$1.accept(Unknown Source)
at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.copyInto(Unknown Source)
at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(Unknown Source)
at java.base/java.util.stream.Nodes$CollectorTask.doLeaf(Unknown Source)
at java.base/java.util.stream.Nodes$CollectorTask.doLeaf(Unknown Source)
at java.base/java.util.stream.AbstractTask.compute(Unknown Source)
at java.base/java.util.concurrent.CountedCompleter.exec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)

OR

2023-10-24 14:50:31,713 ERROR -thread-43 o.a.c.ErrorController      Error for Consume with options {}
java.lang.RuntimeException: Error for Consume with options {}
at org.akhq.utils.Debug.call(Debug.java:74)
at org.akhq.repositories.RecordRepository.consume(RecordRepository.java:133)
at org.akhq.controllers.TopicController.data(TopicController.java:230)
at org.akhq.controllers.$TopicController$Definition$Exec.dispatch(Unknown Source)
at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invoke(AbstractExecutableMethodsDefinition.java:371)
at io.micronaut.context.DefaultBeanContext$4.invoke(DefaultBeanContext.java:594)
at io.micronaut.web.router.AbstractRouteMatch.execute(AbstractRouteMatch.java:303)
at io.micronaut.web.router.RouteMatch.execute(RouteMatch.java:111)
at io.micronaut.http.context.ServerRequestContext.with(ServerRequestContext.java:103)
at io.micronaut.http.server.RouteExecutor.lambda$executeRoute$14(RouteExecutor.java:659)
at reactor.core.publisher.FluxDeferContextual.subscribe(FluxDeferContextual.java:49)
at reactor.core.publisher.InternalFluxOperator.subscribe(InternalFluxOperator.java:62)
at reactor.core.publisher.FluxSubscribeOn$SubscribeOnSubscriber.run(FluxSubscribeOn.java:194)
at io.micronaut.reactive.reactor.instrument.ReactorInstrumentation.lambda$init$0(ReactorInstrumentation.java:62)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
at io.micronaut.scheduling.instrument.InvocationInstrumenterWrappedCallable.call(InvocationInstrumenterWrappedCallable.java:53)
at io.micrometer.core.instrument.composite.CompositeTimer.recordCallable(CompositeTimer.java:129)
at io.micrometer.core.instrument.Timer.lambda$wrap$1(Timer.java:206)
at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.base/java.lang.Thread.run(Unknown Source)
Caused by: java.nio.BufferUnderflowException: null
at java.base/java.nio.Buffer.nextGetIndex(Unknown Source)
at java.base/java.nio.HeapByteBuffer.getShort(Unknown Source)
at org.akhq.utils.ContentUtils.asShort(ContentUtils.java:64)
at org.akhq.utils.ContentUtils.convertToObject(ContentUtils.java:86)
at org.akhq.models.Record.<init>(Record.java:124)
at org.akhq.repositories.RecordRepository.newRecord(RecordRepository.java:491)
at org.akhq.repositories.RecordRepository.consumeOldest(RecordRepository.java:167)
at org.akhq.repositories.RecordRepository.lambda$consume$4(RecordRepository.java:137)
at org.akhq.utils.Debug.call(Debug.java:67)
... 22 common frames omitted

Using the /api/me route I can verify that I have the following permissions (and some additional READ permissions):

• TOPIC: DELETE, READ and READ_CONFIG
• TOPIC_DATA: READ and DELETE
  for topics whose name starts with my principal name (using the pattern "${PRINCIPAL_NAME}-.*" for both permissionsets).

I have not been able to see what kind of topics are affected by the problem: the topic names consist only of lower-case letters, underscore, minus and digits. Some topics differ only in the suffix: v1 instead of v2: one is displayed, the other is not.

Version 0.24 of AKHQ showed the error message "Internal Server Error: Error for Consume with options {}" in the UI if the principal of AKHQ did not have permission to read data from the topic. But now the principal of AKHQ is a superuser, so missing Kafka ACLs can not be a problem.

[AlexisSouquiere]

@BernhardBerbuir you seem to be facing the same issue as the one I faced recently.
Based on the stack trace you put AKHQ has a problem when deserializing some headers in your topic messages (it was headers with unicode on my side).
The PR #1588 will fix the issue. As soon as @tchiotludo will have reviewed it, you shouldn't have any issue

[AlexisSouquiere]

And the fact that you don't see anymore an error message like in 0.24.0 is due to a regression in error management that will be fixed with #1600

[AlexisSouquiere]

@BernhardBerbuir the 2 PR have been merged. Can you try again and tell us if everything is fine now ?

[tchiotludo]

@BernhardBerbuir reopen if not fixed