func New(addrs []string, opts ...KafkaReporterOption) (go2sky.Reporter, error) {
	r := &kafkaReporter{
		c:               sarama.NewConfig(),
		logger:          log.New(os.Stderr, defaultKafkaLogPrefix, log.LstdFlags),
		checkInterval:   defaultCheckInterval,
		topicManagement: defaultTopicManagement,
		topicSegment:    defaultTopicSegment,
	}
	for _, o := range opts {
		o(r)
	}
	p, err := sarama.NewAsyncProducer(addrs, r.c)
	if err != nil {
		return nil, err
	}
	r.producer = p
	if r.c.Producer.Return.Errors {
		go func() {
			for e := range p.Errors() {
				r.logger.Printf("send kafka err: %v", e.Err)
			}
		}()
	}
	return r, nil
}
New create a new reporter to send data to kafka.
package kafkareporter
import (
	"context"
	"log"
	"os"
	"sync"
	"time"
	"github.com/Shopify/sarama"
	"github.com/SkyAPM/go2sky"
	"github.com/SkyAPM/go2sky-plugins/kafkareporter/internal/tool"
	"google.golang.org/protobuf/proto"
	commonv3 "skywalking.apache.org/repo/goapi/collect/common/v3"
	agentv3 "skywalking.apache.org/repo/goapi/collect/language/agent/v3"
	managementv3 "skywalking.apache.org/repo/goapi/collect/management/v3"
)
const (
	defaultCheckInterval   = 20 * time.Second
	defaultKafkaLogPrefix  = "go2sky-kafka"
	topicKeyRegister       = "register-"
	defaultTopicManagement = "skywalking-managements"
	defaultTopicSegment    = "skywalking-segments"
)
type kafkaReporter struct {
	c               *sarama.Config
	producer        sarama.AsyncProducer
	service         string
	serviceInstance string
	wg              sync.WaitGroup
	ctx             context.Context
	cancel          context.CancelFunc
	instanceProps   map[string]string
	logger          *log.Logger
	topicManagement string
	topicSegment    string
	checkInterval   time.Duration
}