func checkConnect(topic string, brokers []string) bool {
	for _, broker := range brokers {
		if c, err := kafka.DialLeader(
			context.Background(),
			"tcp",
			broker,
			topic,
			0); err != nil {
			logx.Error(err)
			return true
		} else if _, err := c.ReadFirstOffset(); err != nil {
			logx.Error(err)
			return true
		}
	}
	return false
}
check kafka connection
package tube
import (
	"context"
	"fmt"
	"io"
	"log"
	"sync"
	"time"
	"github.com/segmentio/kafka-go"
	"github.com/tal-tech/go-zero/core/logx"
)
type partitionMessages struct {
	lastCommitMsg kafka.Message
	commitMsg     kafka.Message
	outputted     map[int64]kafka.Message  
}
type messages struct {
	sync.RWMutex
	data map[int]partitionMessages  
}
type kfkStreamConsumer struct {
	ctx         context.Context
	handleFunc  HandleFunc
	kafkaReader *kafka.Reader
	producer    *kafka.Writer  
	workerNum   int
	pm          messages
	closeFlag   chan struct{}
}
func MustNewKfkStreamConsumer(topic, group string, workerNum int, brokers []string) KfkStreamConsumer {
	done := checkConnect(topic, brokers)
	if done {
		log.Fatal("wrong topic or brokers")
	}
	config := kafka.ReaderConfig{
		Brokers:          brokers,
		GroupID:          group,
		Topic:            topic,
		MinBytes:         10e3,  
		MaxBytes:         10e6,  
		MaxWait:          10 * time.Second,
		SessionTimeout:   10 * time.Second,
		RebalanceTimeout: 5 * 60 * time.Second,
	}
	r := kafka.NewReader(config)
	w := &kafka.Writer{
		Addr:     kafka.TCP(brokers...),
		Topic:    topic + "-failed",
		Balancer: &kafka.LeastBytes{},
		Async:    true,
	}
	ptm := make(map[int]partitionMessages)
	return &kfkStreamConsumer{
		pm:          messages{data: ptm},
		kafkaReader: r,
		producer:    w,
		workerNum:   workerNum,
		closeFlag:   make(chan struct{}),
	}
}