Internal Server Error: Cannot read the array length because "value" is null

[fabiofilz]

Hello,

After producing a tombstone using KafkaNull.INSTANCE AKHQ throws a msg error "Internal Server Error: Cannot read the array length because "value" is null" when I open the main page. If I enter in the topic details to see the content of a message it throws the error "Internal Server Error: Error for Consume with options {}"

I am concerned that if we deploy an application that used KafkaNull.INSTANCE AKHQ will start crashing.

    @Bean
    public Function<Message<CustomerConsentUpdatedEvent>, Message<?>> moveData() {
        return customerConsentUpdated -> {

            String key = Optional.ofNullable(customerConsentUpdated.getHeaders())
                .map(v -> v.get(KafkaHeaders.RECEIVED_MESSAGE_KEY))
                .map(v -> (byte[]) v)
                .map(String::new)
                .get();

            String givesProfilingConsent = Optional.ofNullable(customerConsentUpdated.getPayload().profiling())
                .orElse("NO");

            //  tombstone
            var payLoad = givesProfilingConsent.equals("NO") ? customerConsentUpdated.getPayload() : KafkaNull.INSTANCE; 

            return MessageBuilder
                .withPayload(payLoad)
                .copyHeaders(customerConsentUpdated.getHeaders())
                .setHeader(KafkaHeaders.MESSAGE_KEY, key.getBytes(StandardCharsets.UTF_8))
                .setHeader(KafkaHeaders.TIMESTAMP, customerConsentUpdated.getHeaders().get(KafkaHeaders.RECEIVED_TIMESTAMP))
                .setHeader(KafkaHeaders.TIMESTAMP_TYPE, customerConsentUpdated.getHeaders().get(KafkaHeaders.ORIGINAL_TIMESTAMP_TYPE))
                .build();
        };
    }

The message is produced successfully and I can consume it using the code below or kafka-console-producer:

    @Bean
    public Function<KStream<String, CustomerConsentUpdatedEvent>, KStream<String, CustomerConsentUpdatedEvent>> processTest() {

        return input -> input
            .peek((key, value) -> log.info("Received event key={} value={}", key, Optional.ofNullable(value).map(Objects::toString).orElse("Null")))
            .map((key, value) -> KeyValue.pair(key, value));
    }

kafka.MyTopology : Received event key=123456 value=Null

If I produce the message using the code below, AKHQ doesn't throw the error:

    public static void main(String[] args) {
        String servers = 'localhost:9092';
        String topic = 'tombstone-topic-in';
        String key = '123456';

        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        try(KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            producer.send(new ProducerRecord<String, String>(topic, key, null));
        }

[tchiotludo]

Yes akhq support tombstone and null values.
What KafkaNull.INSTANCE does, I really don't know since I never want to use spring.
To best if to show to difference with kcat WITH -j to output as json between

• the Kafka Stream that work
• the producer with KafkaNull.INSTANCE

or add a debug breakpoint on consumer you try to understand the type of the payload

[fabiofilz]

Hi @tchiotludo ,

Thank you for your response.

I executed the command below:

kcat -b localhost:9092 -C -f '\nKey (%K bytes): %k\t\nValue (%S bytes): %s\n\Partition: %p\tOffset: %o\n--\n' -t customer-consent -J

Sending tombstone using KafkaProducer using Null (Works fine)

{
    "topic": "customer-consent",
    "partition": 0,
    "offset": 0,
    "tstype": "create",
    "ts": 1663683125210,
    "broker": 0,
    "key": "123456",
    "payload": null
}

% Reached end of topic customer-consent [0] at offset 1

Using MessageBuilder with KafkaNull.INSTANCE (Creates the Error)
PS.: I commented out .copyHeaders(customerConsentUpdated.getHeaders()) and the kcat output is the same

Key ({
    "topic": "customer-consent",
    "partition": 0,
    "offset": 1,
    "tstype": "create",
    "ts": 1663683188803,
    "broker": 0,
    "headers": [
        "target-protocol",
        "kafka",
        "contentType",
        "application/json",
        "spring_json_header_types",
        "{\"contentType\":\"java.lang.String\",\"target-protocol\":\"java.lang.String\"}"
    ],
    "key": "123456",
    "payload": null
}

% Reached end of topic customer-consent [0] at offset 2

Thank you.
Fabio

[tchiotludo]

I think I got it, can you provide the full stack trace please for the webserver log ?
I think you have headers with no value, that is not expected

[fabiofilz]

Hi @tchiotludo,

This the error in the console

2022-09-21 14:06:56,593 INFO r-thread-3 org.akhq.log.access [Date: 2022-09-21T14:06:56.102735+01:00] [Duration: 490 ms] [Url: GET /api/dev1-cluster/topic] [Status: 200] [Ip: /0:0:0:0:0:0:0:1] [User: Anonymous]
2022-09-21 14:06:56,795 ERROR r-thread-4 o.a.c.ErrorController Cannot read the array length because "value" is null
java.lang.NullPointerException: Cannot read the array length because "value" is null
at org.akhq.repositories.AvroWireFormatConverter.convertValueToWireFormat(AvroWireFormatConverter.java:39)
at org.akhq.repositories.RecordRepository.newRecord(RecordRepository.java:455)
at org.akhq.repositories.RecordRepository.lambda$getLastRecord$3(RecordRepository.java:109)
at java.base/java.lang.Iterable.forEach(Iterable.java:75)
at org.akhq.repositories.RecordRepository.getLastRecord(RecordRepository.java:107)
at org.akhq.controllers.TopicController.lastRecord(TopicController.java:225)
at org.akhq.controllers.$TopicController$Definition$Exec.dispatch(Unknown Source)
at io.micronaut.context.AbstractExecutableMethodsDefinition$DispatchedExecutableMethod.invoke(AbstractExecutableMethodsDefinition.java:351)
at io.micronaut.context.DefaultBeanContext$4.invoke(DefaultBeanContext.java:583)
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:656)
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$null$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:68)
at io.micrometer.core.instrument.Timer.lambda$wrap$1(Timer.java:171)
at io.micronaut.scheduling.instrument.InvocationInstrumenterWrappedCallable.call(InvocationInstrumenterWrappedCallable.java:53)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
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)
2022-09-21 14:06:56,798 WARN r-thread-3 org.akhq.log.access [Date: 2022-09-21T14:06:56.648272+01:00] [Duration: 150 ms] [Url: GET /api/dev1-cluster/topic/last-record] [Status: 500] [Ip: /0:0:0:0:0:0:0:1] [User: Anonymous]

[fabiofilz]

Hi @tchiotludo.

I hope you are well.

Did you manage to replicate the error? Do you need any further info to identify the error?

Thank you.
Fabio

[tchiotludo]

@fabiofilz : I've added a protection, try with dev in 20 minutes

[fabiofilz]

Working now. Many thanks :-)