Receiving protobuf data triggers a kafka unknown error

[hookover]

version: kafka: 1.1.0 sarama-cluster: 2.1.15 sarama:1.22.1

topic: A, string data eg:

producer
>a
>b 
>c

topic: B, protobuf byte data (cannal read binlog data send to kafka https://github.com/alibaba/canal) eg:

  *��
mysql-bin.000001ǟ�. *UTF-80��ɷ-8B
                                 rupiah_loanJuser_loan_planP�Xb

kafka-console-consumer customer A、B: no any errors topic A customer: no any errors topic B customer errors:

//Every 3 seconds print this: 
Errors kafka: error while consuming rupiah_loan_user_loan/0: kafka server: Unexpected (unknown?) server error.

code

    config := cluster.NewConfig()
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.Initial = sarama.OffsetOldest

    customer, err := cluster.NewConsumer([]string{app.Cfg.App.GetString("KAFKA_BROKER_LIST")}, "dev.test", []string{"credit_data_plus_topic","rupiah_loan_user_loan_plan"}, config)
    if err != nil {
        fmt.Println("kafaka启动失败")
    }
    for {
        select {
        case noti := <-customer.Notifications():
            fmt.Printf("noti: %v\n", noti)
        case err := <-customer.Errors():
            fmt.Printf("errors:%v\n", err)
        case msg := <-customer.Messages():
            fmt.Println(msg.Topic, msg.Offset)
        default:
            time.Sleep(time.Second * 1)
            fmt.Println("间隔1s")
        }
    }

log:

rupiah_loan_user_loan_plan 208
rupiah_loan_user_loan_plan 210
rupiah_loan_user_loan_plan 212
rupiah_loan_user_loan_plan 214
rupiah_loan_user_loan_plan 216
rupiah_loan_user_loan_plan 218
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
rupiah_loan_user_loan_plan 220
rupiah_loan_user_loan_plan 222
rupiah_loan_user_loan_plan 224
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
rupiah_loan_user_loan_plan 226
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
..........

But when I restart the program, it will be consumed from beginning to end. However, if config.Consumer.Offsets.Initial is configured as sarama.OffsetNewest, a piece of data is consumed and the error is always reported. like this

#  go run main.go
rupiah_loan_user_loan_plan 250
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
^Csignal: interrupt

# go run main.go
rupiah_loan_user_loan_plan 260
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.
errors:kafka: error while consuming rupiah_loan_user_loan_plan/0: kafka server: Unexpected (unknown?) server error.

[hookover]

Using other libraries to solve the problem

[wbrucelee]

I also meet this problem