Can I browse messages which are protobuf encoded? #2899

SamsadSajid · 2022-11-07T15:56:37Z

SamsadSajid
Nov 7, 2022

Hi. I have an application that produces protobuf messages to Kafka. I take the messages and serialised them to byte array and publish them to Kafka. When I browse the messages from the Kafka UI it looks as follows:

3 denotes the key

I think this is happening due to some encoding issues. I am wondering if there's a way to browse the protobuf encoded messages.

Answered by Haarolean

Nov 7, 2022

Hi, which version are you on? If you're on latest releast (0.4.0), please try master-labeled docker image first instead. We've implemented serde plugins recently, you can specify which one to use for producing and/or consuming messages.

View full answer

Haarolean · 2022-11-07T15:59:37Z

Haarolean
Nov 7, 2022

Hi, which version are you on? If you're on latest releast (0.4.0), please try master-labeled docker image first instead. We've implemented serde plugins recently, you can specify which one to use for producing and/or consuming messages.

18 replies

SamsadSajid Nov 26, 2022
Author

I was thinking if Kafka-ui can take a file descriptor instead of the original proto file. One of the systems I am currently working with has a proto descriptor file and not the original proto file. i.e. with the following command, a file descriptor is generated from the original proto definition.

protoc -oschema.proto --proto_path=proto/ my_model.proto common.proto

So, the system only knows about the schema.proto which is the file descriptor here. In a test with the kafka-ui, I used this file instead of passing the original proto file i.e.
KAFKA_CLUSTERS_0_PROTOBUFFILES_0: "/schema/schema.proto"

But this failed. Below is the stack trace. I am wondering if kafka-ui does not support it. If not is there any roadmap on whether this will be supported?

ERROR [parallel-1] o.s.b.a.w.r.e.AbstractErrorWebExceptionHandler: [7e5fadcc-35]  500 Server Error for HTTP GET "/api/clusters/local/topic/stream_latency.go/serdes?use=DESERIALIZE"
 java.lang.NullPointerException: Cannot invoke "java.nio.file.Path.toUri()" because "path" is null
       at com.provectus.kafka.ui.serdes.builtin.ProtobufFileSerde.toSchemaDescription(ProtobufFileSerde.java:235)
       Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:
 Error has been observed at the following site(s):
       *__checkpoint ⇢ com.provectus.kafka.ui.config.CustomWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ com.provectus.kafka.ui.config.ReadOnlyModeFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.authorization.AuthorizationWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.authorization.ExceptionTranslationWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.authentication.logout.LogoutWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.savedrequest.ServerRequestCacheWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.context.SecurityContextServerWebExchangeWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.context.ReactorContextWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.header.HttpHeaderWriterWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.config.web.server.ServerHttpSecurity$ServerWebExchangeReactorContextWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.security.web.server.WebFilterChainProxy [DefaultWebFilterChain]
       *__checkpoint ⇢ org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter [DefaultWebFilterChain]
       *__checkpoint ⇢ HTTP GET "/api/clusters/local/topic/stream_latency.go/serdes?use=DESERIALIZE" [ExceptionHandlingWebHandler]
 Original Stack Trace:
               at com.provectus.kafka.ui.serdes.builtin.ProtobufFileSerde.toSchemaDescription(ProtobufFileSerde.java:235)
               at java.base/java.util.Optional.map(Optional.java:260)
               at com.provectus.kafka.ui.serdes.builtin.ProtobufFileSerde.getSchema(ProtobufFileSerde.java:229)
               at com.provectus.kafka.ui.serdes.SerdeInstance.lambda$getSchema$0(SerdeInstance.java:43)
               at com.provectus.kafka.ui.serdes.SerdeInstance.wrapWithClassloader(SerdeInstance.java:32)
               at com.provectus.kafka.ui.serdes.SerdeInstance.getSchema(SerdeInstance.java:43)
               at com.provectus.kafka.ui.service.DeserializationService.toDto(DeserializationService.java:131)
               at com.provectus.kafka.ui.service.DeserializationService.getSerdesForDeserialize(DeserializationService.java:119)
               at com.provectus.kafka.ui.controller.MessagesController.getSerdes(MessagesController.java:137)
               at com.provectus.kafka.ui.controller.MessagesController$$FastClassBySpringCGLIB$$8951e2d8.invoke(<generated>)
               at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
               at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:793)
               at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
               at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
               at org.springframework.validation.beanvalidation.MethodValidationInterceptor.invoke(MethodValidationInterceptor.java:123)
               at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
               at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:763)
               at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:708)
               at com.provectus.kafka.ui.controller.MessagesController$$EnhancerBySpringCGLIB$$f7dc8f37.getSerdes(<generated>)
               at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
               at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
               at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
               at java.base/java.lang.reflect.Method.invoke(Method.java:568)
               at org.springframework.web.reactive.result.method.InvocableHandlerMethod.lambda$invoke$0(InvocableHandlerMethod.java:144)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:125)
               at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
               at reactor.core.publisher.MonoZip$ZipCoordinator.signal(MonoZip.java:251)
               at reactor.core.publisher.MonoZip$ZipInner.onNext(MonoZip.java:336)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:180)
               at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2398)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.request(MonoPeekTerminal.java:139)
               at reactor.core.publisher.MonoZip$ZipInner.onSubscribe(MonoZip.java:325)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onSubscribe(MonoPeekTerminal.java:152)
               at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:55)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:240)
               at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.onComplete(MonoIgnoreThen.java:203)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onComplete(MonoFlatMap.java:181)
               at reactor.core.publisher.Operators.complete(Operators.java:137)
               at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:120)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:263)
               at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:74)
               at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:82)
               at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:282)
               at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:863)
               at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:180)
               at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2398)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.request(MonoPeekTerminal.java:139)
               at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:171)
               at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2194)
               at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:2068)
               at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:96)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onSubscribe(MonoPeekTerminal.java:152)
               at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:55)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:451)
               at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:219)
               at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:165)
               at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:87)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:82)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:299)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onComplete(MonoPeekTerminal.java:299)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:148)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:74)
               at reactor.core.publisher.FluxFilter$FilterSubscriber.onNext(FluxFilter.java:113)
               at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onNext(FluxDefaultIfEmpty.java:101)
               at reactor.core.publisher.MonoNext$NextSubscriber.onNext(MonoNext.java:82)
               at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.innerNext(FluxConcatMap.java:282)
               at reactor.core.publisher.FluxConcatMap$ConcatMapInner.onNext(FluxConcatMap.java:863)
               at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:151)
               at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:129)
               at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onNext(FluxFilterFuseable.java:118)
               at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2398)
               at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.request(FluxFilterFuseable.java:191)
               at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:171)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onSubscribe(MonoFlatMap.java:110)
               at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:96)
               at reactor.core.publisher.FluxFilterFuseable$FilterFuseableSubscriber.onSubscribe(FluxFilterFuseable.java:87)
               at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:55)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.drain(FluxConcatMap.java:451)
               at reactor.core.publisher.FluxConcatMap$ConcatMapImmediate.onSubscribe(FluxConcatMap.java:219)
               at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:165)
               at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:87)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.MonoIgnoreThen$ThenIgnoreMain.subscribeNext(MonoIgnoreThen.java:263)
               at reactor.core.publisher.MonoIgnoreThen.subscribe(MonoIgnoreThen.java:51)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:82)
               at reactor.core.publisher.FluxFilter$FilterSubscriber.onComplete(FluxFilter.java:166)
               at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onComplete(FluxPeekFuseable.java:940)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:85)
               at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2400)
               at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.set(Operators.java:2194)
               at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.onSubscribe(Operators.java:2068)
               at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:55)
               at reactor.core.publisher.Mono.subscribe(Mono.java:4400)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onComplete(FluxSwitchIfEmpty.java:82)
               at reactor.core.publisher.MonoNext$NextSubscriber.onComplete(MonoNext.java:102)
               at reactor.core.publisher.FluxFilter$FilterSubscriber.onComplete(FluxFilter.java:166)
               at reactor.core.publisher.FluxFlatMap$FlatMapMain.checkTerminated(FluxFlatMap.java:846)
               at reactor.core.publisher.FluxFlatMap$FlatMapMain.drainLoop(FluxFlatMap.java:608)
               at reactor.core.publisher.FluxFlatMap$FlatMapMain.drain(FluxFlatMap.java:588)
               at reactor.core.publisher.FluxFlatMap$FlatMapMain.onComplete(FluxFlatMap.java:465)
               at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onComplete(FluxPeekFuseable.java:277)
               at reactor.core.publisher.FluxIterable$IterableSubscription.slowPath(FluxIterable.java:294)
               at reactor.core.publisher.FluxIterable$IterableSubscription.request(FluxIterable.java:230)
               at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:144)
               at reactor.core.publisher.FluxFlatMap$FlatMapMain.onSubscribe(FluxFlatMap.java:371)
               at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onSubscribe(FluxPeekFuseable.java:178)
               at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:165)
               at reactor.core.publisher.FluxIterable.subscribe(FluxIterable.java:87)
               at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:64)
               at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52)
               at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157)
               at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1816)
               at reactor.core.publisher.FluxDefaultIfEmpty$DefaultIfEmptySubscriber.onComplete(FluxDefaultIfEmpty.java:109)
               at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
               at reactor.core.publisher.FluxMap$MapSubscriber.onComplete(FluxMap.java:144)
               at reactor.core.publisher.FluxFilter$FilterSubscriber.onComplete(FluxFilter.java:166)
               at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onComplete(FluxMap.java:275)
               at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1817)
               at reactor.core.publisher.MonoCacheTime$CoordinatorSubscriber.signalCached(MonoCacheTime.java:337)
               at reactor.core.publisher.MonoCacheTime$CoordinatorSubscriber.onNext(MonoCacheTime.java:354)
               at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:200)
               at reactor.core.publisher.FluxSwitchIfEmpty$SwitchIfEmptySubscriber.onNext(FluxSwitchIfEmpty.java:74)
               at reactor.core.publisher.MonoPeekTerminal$MonoTerminalPeekSubscriber.onNext(MonoPeekTerminal.java:180)
               at reactor.core.publisher.MonoPublishOn$PublishOnSubscriber.run(MonoPublishOn.java:181)
               at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
               at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
               at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
               at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
               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)

SamsadSajid Nov 26, 2022
Author

And if I have multiple topics, I need to pass multiple types in the docker-compose. How do I do that? This links illustrates how to do it but likewise no mention of the docker-compose.

Currently, I'm passing only one type:

KAFKA_CLUSTERS_0_PROTOBUFMESSAGENAME: "proto.MyTopicModel"

But I reckon I need a map of topics to type since I have multiple types for multiple topics. How do I pass the map for the config protobufMessageNameByTopic in the docker-compose?

Haarolean Apr 14, 2023

@SamsadSajid https://docs.kafka-ui.provectus.io/configuration/serialization-serde

SamsadSajid Apr 14, 2023
Author

Thanks, @Haarolean. The documentation is very good. 🚀

Btw, do you folks have any roadmap on supporting proto file descriptors instead of the actual proto files that we pass in the protobufFilesDir config?

Haarolean Apr 17, 2023

@SamsadSajid please raise an issue

amymyo · 2022-11-29T05:47:11Z

amymyo
Nov 29, 2022

I have uploaded the protobuf schema to schema registry, but still cannot browse the messages either, I'm on master and schema is in protobuf 2. Is there a way to browse the message with protobuf schema?

1 reply

Haarolean Dec 2, 2022

@amymyo please raise a new issue/discussion with your schema example and other stuff asked in an issue template

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Can I browse messages which are protobuf encoded? #2899

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 2 comments 19 replies

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Select a reply

Uh oh!

Can I browse messages which are protobuf encoded? #2899

Uh oh!

Uh oh!

SamsadSajid Nov 7, 2022

Replies: 2 comments · 19 replies

Uh oh!

Haarolean Nov 7, 2022

Uh oh!

SamsadSajid Nov 26, 2022 Author

Uh oh!

SamsadSajid Nov 26, 2022 Author

Uh oh!

Haarolean Apr 14, 2023

Uh oh!

SamsadSajid Apr 14, 2023 Author

Uh oh!

Haarolean Apr 17, 2023

Uh oh!

amymyo Nov 29, 2022

Uh oh!

Haarolean Dec 2, 2022

SamsadSajid
Nov 7, 2022

Replies: 2 comments 19 replies

Haarolean
Nov 7, 2022

SamsadSajid Nov 26, 2022
Author

SamsadSajid Nov 26, 2022
Author

SamsadSajid Apr 14, 2023
Author

amymyo
Nov 29, 2022