protobuf ingestion support

[MattUebel]

being able to read in records in the protobuf format would be a great feature for us

[chaitanyaphalak]

Hey @MattUebel can you explain more?
Are you talking about adding support for io.confluent.connect.protobuf.ProtobufConverter ?
You can set that on the worker right? Are you seeing any errors?

[MattUebel]

Essentially, the ability to have the connector consume protobuf encoded topics.

This is already supported? And just a matter of configuring the worker itself with the appropriate schema to decode the protobufs?

[chaitanyaphalak]

oh I guess you mean whether the connector natively deserializes protobuf messages?
The connector doesn't support that, yet.

[MattUebel]

oh I guess you mean whether the connector natively deserializes protobuf messages?

Yup! That sounds about right!

Would an ideas.splunk.com and/or support P4 help prioritize this? It would be a huge boon to us 😄

[chaitanyaphalak]

@MattUebel contacting support will definitely work :)

[MattUebel]

👋 @chaitanyaphalak , I created support case #1858946 to go along with this issue!

[bluedog13]

The Confluent managed version of the connector is able to read data from topics that are serialized using PROTOBUF.

[elvis-cai]

[2022-03-08 11:21:22,542] ERROR WorkerSinkTask{id=kafka-connect-splunk-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
org.apache.kafka.connect.errors.ConnectException: Tolerance exceeded in error handler
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:206)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execute(RetryWithToleranceOperator.java:132)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.convertAndTransformRecord(WorkerSinkTask.java:495)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.convertMessages(WorkerSinkTask.java:475)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:325)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:229)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:201)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:185)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.connect.errors.DataException: Failed to deserialize data for topic test-topic to Avro: 
	at io.confluent.connect.avro.AvroConverter.toConnectData(AvroConverter.java:118)
	at org.apache.kafka.connect.storage.Converter.toConnectData(Converter.java:87)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.lambda$convertAndTransformRecord$0(WorkerSinkTask.java:495)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndRetry(RetryWithToleranceOperator.java:156)
	at org.apache.kafka.connect.runtime.errors.RetryWithToleranceOperator.execAndHandleError(RetryWithToleranceOperator.java:190)
	... 13 more
Caused by: org.apache.kafka.common.errors.SerializationException: Unknown magic byte!

Get the same issue here for protobuf topic, keen to get supported for protobuf topic.

[amitjain-wk]

1858946

Can somebody point me to the documentation for PROTObuf scenario

[foram-splunk]

This issue is resolved in #341