func KafkaMessageAsZerologObject(message kafka.Message) log.LoggerFunc {
	return func(e *zerolog.Event) {
		e.
			Str("topic", message.Topic).
			Int64("offset", message.Offset).
			Time("time", message.Time).
			Int("size", len(message.Value))
	}
}
KafkaMessageAsZerologObject converts a kafka message into a LogObjectMarshaler.
package consumer
import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/url"
	"strings"
	"github.com/ccamel/go-graphql-subscription-example/server/log"
	"github.com/ccamel/go-graphql-subscription-example/server/source"
	"github.com/reactivex/rxgo/v2"
	"github.com/rs/zerolog"
	uuid "github.com/satori/go.uuid"
	"github.com/segmentio/kafka-go"
)
var ErrNoBrokerSpecified = errors.New("no broker specified")
type kafkaSource struct {
	uri     *url.URL
	brokers []string
}
func (s kafkaSource) URI() *url.URL {
	return s.uri
}
type kafkaConsumer struct {
	consumerID uuid.UUID
	brokers    []string
	topic      string
	offset     int64
	ctx        context.Context
	log        zerolog.Logger
}
func newKafkaSource(uri *url.URL) (source.Source, error) {
	brokers, err := parseKafkaBrokers(uri)
	if err != nil {
		return nil, err
	}
	return &kafkaSource{
		uri:     uri,
		brokers: brokers,
	}, nil
}
func (s kafkaSource) NewConsumer(ctx context.Context, topic string, offset int64) rxgo.Observable {
	id := uuid.NewV4()
	c := kafkaConsumer{
		id,
		s.brokers,
		topic,
		offset,
		ctx,
		zerolog.
			Ctx(ctx).
			With().
			Str("consumerID", id.String()).
			Logger(),
	}
	return makeObservableFromKafkaConsumer(c)
}