func unmarshalKafkaMessage(c kafkaConsumer, m kafka.Message) (map[string]interface{}, bool) {
	var v map[string]interface{}
	if err := json.Unmarshal(m.Value, &v); err != nil {
		c.log.
			Warn().
			Object("message", KafkaMessageAsZerologObject(m)).
			Err(err).
			Msg("⚱️ Failed to unmarshal message (message will be dropped)")
		return nil, false
	}
	return v, true
}
unmarshalKafkaMessage parses the given kafka message into a JSON map.
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)
}
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))
	}
}
func parseKafkaBrokers(source *url.URL) ([]string, error) {
	brokersStr := source.Query().Get("brokers")
	if brokersStr == "" {
		return nil, fmt.Errorf("incorrect source URL %s: %w", source.String(), ErrNoBrokerSpecified)
	}
	return strings.Split(brokersStr, ","), nil
}