func Start(ctx context.Context, tp config.Tracepoint, bufpool *sync.Pool, ch chan *bytes.Buffer) error {
	var (
		cfg  = config.FromContext(ctx)
		kCfg = kafkaConfig(cfg.Egress[tp.Egress].Config)
	)
	sCfg, err := saramaConfig(kCfg)
	if err != nil {
		return err
	}
	err = sCfg.Validate()
	if err != nil {
		return err
	}
	k := kafka{
		bufpool: bufpool,
		dCh:     ch,
	}
	k.producer, err = sarama.NewAsyncProducer(kCfg.Brokers, sCfg)
	if err != nil {
		return err
	}
	k.hostname()
	switch kCfg.Serialization {
	case "spb":
		k.bCh = make(chan []byte, 1000)
		for i := 0; i < kCfg.Workers; i++ {
			go k.workerSPB(ctx, cfg.Fields[tp.Fields])
		}
		k.protobufLoop(ctx, kCfg.Topic)
	case "pb":
		k.bCh = make(chan []byte, 1000)
		for i := 0; i < kCfg.Workers; i++ {
			go k.workerPB(ctx, cfg.Fields[tp.Fields])
		}
		k.protobufLoop(ctx, kCfg.Topic)
	case "json":
		k.jsonLoop(ctx, kCfg.Topic)
	}
	return nil
}
Start starts producing the requested fields to kafka cluster.
package kafka
import (
	"bytes"
	"context"
	"fmt"
	"os"
	"sync"
	"github.com/Shopify/sarama"
	"go.uber.org/zap"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"github.com/mehrdadrad/tcpdog/config"
	"github.com/mehrdadrad/tcpdog/egress/helper"
	pb "github.com/mehrdadrad/tcpdog/proto"
)
type kafka struct {
	producer sarama.AsyncProducer
	bufpool  *sync.Pool
	dCh      chan *bytes.Buffer
	bCh      chan []byte
	jsonTail []byte
}