[kafka]: unmarshal message failed: failed to unmarshal, message is *broker.Any, want proto.Message

yimuczhou commented 1 year ago

按照Kraots示例跑了下，出现这个问题，一般是什么原因导致

yimuczhou commented 1 year ago

`func registerKafkaSubscribers(ctx context.Context, srv kafka.Server, svc service.CanalService) { registerInsertSensorDataHandler(ctx, srv, svc) }

type HygrothermographHandler func(_ context.Context, topic string, headers broker.Headers, msg *v1.CanalData) error

func handleHygrothermograph(_ context.Context, topic string, headers broker.Headers, msg *v1.CanalData) error { log.Infof("Topic %s, Headers: %+v, Payload: %+v\n", topic, headers, msg) return nil }

func RegisterHygrothermographJsonHandler(fnc HygrothermographHandler) broker.Handler { return func(ctx context.Context, event broker.Event) error { var msg v1.CanalData switch t := event.Message().Body.(type) { case []byte: msg = &v1.CanalData{} if err := json.Unmarshal(t, msg); err != nil { return err } fmt.Println(msg.String()) case v1.CanalData: if err := fnc(ctx, event.Topic(), event.Message().Headers, t); err != nil { return err } default: return fmt.Errorf("unsupported type: %T", t) } return nil } }

func registerInsertSensorDataHandler(ctx context.Context, srv kafka.Server, svc service.CanalService) { //binder := func() broker.Any { return &v1.CanalData{} }

_ = srv.RegisterSubscriber(ctx,
    "php-device-beta",
    "kratos",
    false,
    RegisterHygrothermographJsonHandler(handleHygrothermograph),
    nil,
)

}`

帮binder注释掉还是会报