Skip to content

draft: Zfmt refactor #31

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ethaneason wants to merge 3 commits into
base: main
Choose a base branch
from
Open

draft: Zfmt refactor #31

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
baf9f20
remove dependency on zfmt
Sep 12, 2025
755bf6e
initial refactor
Sep 19, 2025
04b9671
fix tests
Jan 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 7 additions & 59 deletions client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,8 @@ package zkafka

import (
"context"
"fmt"
"sync"

"github.com/zillow/zfmt"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/trace"
)
Expand Down Expand Up @@ -36,15 +34,12 @@ type Client struct {
tp trace.TracerProvider
p propagation.TextMapPropagator

srf *schemaRegistryFactory

producerProvider confluentProducerProvider
consumerProvider confluentConsumerProvider
}

// NewClient instantiates a kafka client to get readers and writers
func NewClient(conf Config, opts ...Option) *Client {
srf := newSchemaRegistryFactory()
c := &Client{
conf: conf,
readers: make(map[string]*KReader),
Expand All @@ -53,7 +48,6 @@ func NewClient(conf Config, opts ...Option) *Client {

producerProvider: defaultConfluentProducerProvider{}.NewProducer,
consumerProvider: defaultConfluentConsumerProvider{}.NewConsumer,
srf: srf,
}
for _, opt := range opts {
opt(c)
Expand Down Expand Up @@ -81,20 +75,15 @@ func (c *Client) Reader(_ context.Context, topicConfig ConsumerTopicConfig, opts
if exist && !r.isClosed {
return r, nil
}

formatter, err := c.getFormatter(formatterArgs{
formatter: topicConfig.Formatter,
schemaID: topicConfig.SchemaID,
srCfg: topicConfig.SchemaRegistry,
})
marshaler, err := topicConfig.MarshalerFactory.GetMarshaler()
if err != nil {
return nil, err
}
reader, err := newReader(readerArgs{
cfg: c.conf,
cCfg: topicConfig,
consumerProvider: c.consumerProvider,
f: formatter,
marshaler: marshaler,
l: c.logger,
prefix: c.groupPrefix,
hooks: c.lifecycle,
Expand Down Expand Up @@ -128,20 +117,15 @@ func (c *Client) Writer(_ context.Context, topicConfig ProducerTopicConfig, opts
if exist && !w.isClosed {
return w, nil
}
formatter, err := c.getFormatter(formatterArgs{
formatter: topicConfig.Formatter,
schemaID: topicConfig.SchemaID,
srCfg: topicConfig.SchemaRegistry,
})

marshaler, err := topicConfig.MarshalerFactory.GetMarshaler()
if err != nil {
return nil, err
}
writer, err := newWriter(writerArgs{
cfg: c.conf,
pCfg: topicConfig,
producerProvider: c.producerProvider,
f: formatter,
marshaler: marshaler,
l: c.logger,
t: getTracer(c.tp),
p: c.p,
Expand Down Expand Up @@ -176,42 +160,6 @@ func (c *Client) Close() error {
return err
}

func (c *Client) getFormatter(args formatterArgs) (kFormatter, error) {
formatter := args.formatter
schemaID := args.schemaID

switch formatter {
case AvroSchemaRegistry:
scl, err := c.srf.createAvro(args.srCfg)
if err != nil {
return nil, err
}
return newAvroSchemaRegistryFormatter(scl), nil
case ProtoSchemaRegistry:
scl, err := c.srf.createProto(args.srCfg)
if err != nil {
return nil, err
}
cf := newProtoSchemaRegistryFormatter(scl)
return cf, nil
case JSONSchemaRegistry:
scl, err := c.srf.createJson(args.srCfg)
if err != nil {
return nil, err
}
cf := newJsonSchemaRegistryFormatter(scl)
return cf, nil
case CustomFmt:
return &errFormatter{}, nil
default:
f, err := zfmt.GetFormatter(formatter, schemaID)
if err != nil {
return nil, fmt.Errorf("unsupported formatter %s", formatter)
}
return zfmtShim{F: f}, nil
}
}

func getTracer(tp trace.TracerProvider) trace.Tracer {
if tp == nil {
return nil
Expand All @@ -223,8 +171,8 @@ func getWriterKey(cfg ProducerTopicConfig) string {
return cfg.ClientID + "-" + cfg.Topic
}

type formatterArgs struct {
formatter zfmt.FormatterType
type marshalerArgs struct {
marshaler KMarshaler
schemaID int
srCfg SchemaRegistryConfig
rCfg SchemaRegistryConfig
}
Loading
Loading