Confluent-kafka-go: Schema registry support

[skyrocknroll]

I am not able to pass url for schema registry url in the config map. It would be great if somebody point me to right place. I looked at all the documentation. We are using confluent kakfa and schema registry all over the place. In java & python we are able to pass. But not able to use it go.

[ewencp]

@skyrocknroll At the moment, Avro support has not yet been integrated into the Go client. Messages will contain byte[] data which you should serialize/deserialize yourself.

We'll be looking to integrate Avro serialization and schema registry support with the client soon.

[skyrocknroll]

@ewencp Thank you for your response. Consumer we can ignore first 4 bytes. But In Producer we need to get the schema id. Right now i see one workaround as directly hitting the schema registry and generating the schema id and embed it in message. is it right and Is there any other better work around ? Also if you can help me on ETA it would be great. It would be great if you point me to a place where i can implement and raise an pull request.

[lingzijordan]

yeah, facing the same issue now. Go client can't use schema registry which is sad...

[thoughtpoet]

I see I am not alone. I have been using the schema registry a good while with Kafka Streams and the Scala DSL. Now I need to incorporate python, which seems fine, and Go, which lands me here. What has been the best workaround to date? Prepend a magic byte followed by 16-byte MD5 hash?

[thoughtpoet]

Problem sorted. Somewhere I'd read about needing a 16-byte MD5 hash. Instead it's a magic byte of 0x00 followed byte the schema id, which is presumably an unsigned integer of four bytes. So, since this particular schema in the registry has {"id": 1}, the following header worked:

    bs := make([]byte, 4)
    binary.BigEndian.PutUint32(bs, 1)
    hdr = append([]byte{0}, bs...)

Using goavro, my code looked like this:
deliveryChan = make(chan kafka.Event)
topic := "v1-test-create-user"
err = producer.Produce(
&kafka.Message{
TopicPartition: kafka.TopicPartition{
Topic: &topic,
Partition: kafka.PartitionAny},
Value: append(hdr, b...)},
deliveryChan)

It's not much additional work, just took some sleuthing to understand what to do.

[ewencp]

For anyone else wondering about this, the format is documented here.

[skyrocknroll]

@ewencp Any updates here ? An rough ETA would be great on our side.

[thoughtpoet]

Likewise! -- Fred Patton

On Mon, Aug 14, 2017 at 7:37 AM Yuvaraj [e-mail] wrote:

@ewencp https://github.com/ewencp Any updates here ? An rough ETA would
be great on our side.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
https://github.com/confluentinc/confluent-kafka-go/issues/69#issuecomment-322207687,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AHt7T_y1PaxjU0j22NdWGTyWr3HkZNYRks5sYFuVgaJpZM4N5b-1
.

[ewencp]

Sorry, we don't have specific release we can tie this feature to yet.

[JasonSanchez]

👍

[larsp]

Are there any news regarding schema registry?

[edenhill]

Aiming for Q2 or early Q3

[peoplemerge]

Looking forward to seeing this too. 👍

[skyrocknroll]

@edenhill Is thill still on priority?

[edenhill]

Yes, still scheduled for Q3

[lukethomas1]

@edenhill Any further updates?

[edenhill]

@lukethomas1 We're aiming to add Schema-Registry and Avro serializer support in Q3 or Q4.
cc @rnpridgeon

[avinashramesh91]

@edenhill Is this still on for Q3 ? We have a lot of hopes for this support to arrive soon.

[edenhill]

Yes, or early Q4. cc @rnpridgeon

[avinashramesh91]

It would be great if you could specifically mention Q3 or Q4 since our choice of confluent-kafka-go for many microservices in my company depends on this! Thanks

[avinashramesh91]

Hi @edenhill

I see the issue is still open. So the schema registry support is still not there? Is it on the plans for the near future? Thanks in advance!

[edenhill]

We're looking to add a schema-registry client during the coming month.
See this PR: https://github.com/confluentinc/confluent-kafka-go/pull/231

[PremchandN]

I am also looking forward to this feature in Go. Thanks in advance.

[cobolbaby]

+1

[lucasmarfe]

+1

[diepmv]

+1

[bernardoVale]

+1

[apabla]

Is there a golang library available with support for a schema registry URL? My current kafka.ConfigMap config will not allow "schema.registry.url"

Could anyone advise on other kafka client libraries to use?

Running go
panic: No such configuration property: "schema.registry.url"

[eko]

+1 for this

[supasonk]

+1 for this too, can anyone give us an eta on the release date?

[edenhill]

Current plan is to start looking into this again in Q3 of 2020.

[james-johnston-thumbtack]

It seems like one potential issue might be lack of schema evolution support in the commonly-used goavro package: https://github.com/linkedin/goavro/issues/166

[mhowlett]

@james-johnston-thumbtack - with go, we'll do protobuf / json first. we might not do avro if there are feature gaps (from what I know, and from what you say, this is the case).

[sneko]

Hi @edenhill ,

I always used JSON data into Kafka and I'm looking for switching over Avro to respect contracts and avoid weird behavior when something changes.

According to this thread, some things will be embedded into this library but I'm not sure how far it will manage everything.

For now, before reading this thread I thought about:

• Retrieve schemas/subjects thanks to https://github.com/lensesio/schema-registry
• Instantiate a codec (encode/decode) thanks to https://github.com/linkedin/goavro (but some others exist, even if they don't manage logical types, or anonymous fields...)
• Consume/produce messages with confluent-kafka-go and encode/decode with the above codec

But when looking at comments, no one mentioned a working way, is it possible with what I mentioned? If no, can someone explain why? And does the PR planned for Q3 will handle everything?

BTW, for my service-to-service communication I use Protobuf, already familiar with it, I know Protobuf still have a long path to be working in Kafka tools like KSQL... but I'm wondering if it would not be better to use it over Avro? What do you think?

Thank you in advance, having your thoughts will help me :)

[mhowlett]

we are unlikely to support avro in the go client due to features of available libraries (i'm not familiar with the options myself, but I know this is our current stance).

we do plan to support JSON and Protobuf, it just hasn't been prioritized yet (it is relatively high priority though). Both protobuf and json now have first class support across confluent platform (to my knowledge), including KSQL.

If you're familiar with protobuf, i'd highly recommend using that over avro due to superior library implementation.

[apabla]

Just to share with folks on this subject. I have been working on a conversion for Kafka consumers to golang. The current golang confluent library does not support a configuration used in the java consumer- "schema.registry.url". Other libraries that I have considered linkdin, sarma hardcode the schema in the go consumer. I need to support dynamic schema versions. I have been reviewing the implementtion provided by https://github.com/dangkaka
https://github.com/dangkaka/go-kafka-avro/blob/e714c259f1e9b42b05cc03ca98ed9b476a970dff/schemaRegistry.go#L66 I am working though his solution to see if this is a possibility. Hope this helps others dealing with this problem.

[apabla]

I was struggling with implementing a golang Kafka Avro consumer that supported TLS connectivity. Hope the solution below assists others. Thank you to Alexander Lokshin and Rakeshreddy Kotha for providing assistance with arriving at this solution.

`
import (
"crypto/tls"
"crypto/x509"
"encoding/binary"
"encoding/json"
"fmt"
"io/ioutil"
"log"
"os"
"testing"
"time"

"github.com/Shopify/sarama"
"github.com/confluentinc/confluent-kafka-go/kafka"
"github.com/riferrei/srclient"
"github.com/stretchr/testify/assert"
"github.com/tidwall/pretty"

)

var (
producerAvro *kafka.Producer
consumerAvro *kafka.Consumer
avroschemaregclient *srclient.SchemaRegistryClient
hostbootstrap = []string{"YOUR BOOTSTRAP ENTRIES"}

clientCert = "/pathto/client.cer.pem"
clientKey  = "/pathto/client.key.pem"
caCert     = "/pathto/server.cer.pem"

)

func TestConsumerGoAvro(t *testing.T) {
//configure TLSConfig component - Ref Mickaël Rémond blog
//https://www.process-one.net/blog/using-tls-authentication-for-your-go-kafka-client/
tlsConfig, err := TLSConfig(clientCert, clientKey, caCert)
if err != nil {
log.Fatalf("error: %q", err)
}
//Configure Kafka consumer. Note the tlsConfig on line 39
consumerConfig := sarama.NewConfig()
consumerConfig.Net.TLS.Enable = true
consumerConfig.Net.TLS.Config = tlsConfig
consumerConfig.Net.KeepAlive = 30 * time.Second
consumerConfig.Consumer.Retry.Backoff = 25 * time.Millisecond

sarama.Logger = log.New(os.Stdout, "[anuanuanu] ", log.LstdFlags)

//connect to your kafka cluster
client, err := sarama.NewClient(hostbootstrap, consumerConfig)
if err != nil {
    log.Fatalf("Kafka client connection failure: %q", err)
}
consumerAvro, err := sarama.NewConsumerFromClient(client)
if err != nil {
    log.Fatal(err)
}
//Test to confirm consumer is consturcted
assert.NotNil(t, consumerAvro, "Created Consumer")

//Create a schema reg client
avroschemaregclient := srclient.CreateSchemaRegistryClient("http://localhost:8081")
for {
    //Read data from a single topic,partiton,offset
    partitionConsumer, err := consumerAvro.ConsumePartition("NAMEOFYOURTOPIC", 0, 0)
    if err != nil {
        log.Println(err)
    }
    consumed := 0
    // I have access to my schema list and hard coded the id for this test http://localhost:8081/subjects/TOPICNAME/versions/latest
    schemaID := 56
    schema, err := avroschemaregclient.GetSchema(int(schemaID))
    //Consume data from the kafka topic
ConsumerLoop:
    for {
        msg := <-partitionConsumer.Messages()
        log.Printf("Consumed message key %s\n", msg.Key)
        log.Printf("Consumed message value %s\n", msg.Value)
        log.Printf("Consumed message offset %d\nData:", msg.Offset)
        native, _, _ := schema.Codec().NativeFromBinary(msg.Value[5:])
        value, _ := schema.Codec().TextualFromNative(nil, native)
        fmt.Printf("Value deserialized %s\n", pretty.Pretty(value))
        consumed++
        if consumed == 3 {
            break ConsumerLoop
        }
    }
    break
}
consumerAvro.Close()

}

func TLSConfig(clientCertFile, clientKeyFile, caCertFile string) (*tls.Config, error) {
tlsConfig := tls.Config{}

// Load client cert
cert, err := tls.LoadX509KeyPair(clientCertFile, clientKeyFile)
if err != nil {
    return &tlsConfig, err
}
tlsConfig.Certificates = []tls.Certificate{cert}

// Load CA cert
caCert, err := ioutil.ReadFile(caCertFile)
if err != nil {
    return &tlsConfig, err
}
caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)
tlsConfig.RootCAs = caCertPool

tlsConfig.BuildNameToCertificate()
return &tlsConfig, err

}

`

[alok87]

You can try this https://github.com/riferrei/srclient, it works.

[function1983]

You can try this https://github.com/riferrei/srclient, it works.

I can confirm that it works on my Kafka stack (tested with avro).

[briansorahan]

@mhowlett I'm curious how protobuf will integrate with Go-based schema registry clients since Go is typically deployed as a statically-linked binary and protobuf Go support depends on code generation.
Do you have any juicy details you could share?

[shivakumarss]

+1

[shivakumarss]

I tried using mentioned api for schema creation, and used kafka go client. It works in self managed cluster but doesnt seem to work in confluent cloud. Please have a look at this ticket and suggest.
https://github.com/riferrei/srclient/issues/15

Any help would be really appreciated. I am just stuck.

[gklijs]

You might want to check https://crates.io/crates/schema_registry_converter there are some issues with complex Avro schema's, because of the used library for Avro, and for protobuf it's doing 'just' the schema registry stuff. Could be used as inspiration, or might be compiled and wrapped?