Hey, I am trying to configure Kafka to consume cloudevents. Seems like I was getting errors at kafka initialization.

{"json":{"src":"base.go:73"},"level":"fatal","msg":"caught panic: kafka: client has run out of available brokers to talk to (Is your cluster reachable?) [goroutine 1 [running]:\nruntime/debug.Stack()\n\t/usr/local/go/src/runtime/debug/stack.go:24 +0x65\<http://ngithub.com/flyteorg/flyteadmin/pkg/rpc/adminservice.NewAdminServer.func1()\n\t/go/src/github.com/flyteorg/flyteadmin/pkg/rpc/adminservice/base.go:73|ngithub.com/flyteorg/flyteadmin/pkg/rpc/adminservice.NewAdminServer.func1()\n\t/go/src/github.com/flyteorg/flyteadmin/pkg/rpc/adminservice/base.go:73> +0x88\npanic({0x224d580, 0xc0009c8300})\n\t/usr/local/go/src/runtime/panic.go:838 +0x207\<http://ngithub.com/flyteorg/flyteadmin/pkg/async/cloudevent.NewCloudEventsPublisher(|ngithub.com/flyteorg/flyteadmin/pkg/async/cloudevent.NewCloudEventsPublisher(>{0x2bd10a0, 0xc00005a018}, {0x1, {0xc000f2e460, 0x5}, {{0x0, 0x0}},

The weird thing is that we tried to use the same version sarama client to connect to the kafka instance in the same cluster, and it works. （Also I was able to success at this step, yet after I redeployed with eventTypes set to a single type it failed and could never get back to normal) The cloudevent config I used is the following：

cloudevents:
    enable: true
    kafka:
      brokers: kafkaip
      version:
        version:
          - 2
          - 2
          - 0
          - 0
    eventsPublisher:
      eventTypes:
      - all
      topicName: workflow-engine-test
    type: Kafka

Any clue what could be the problem?

Hey @Dan Rammer (hamersaw). Would you be able to help?

add up to that, the error was pretty flaky. I was able to create couple deployment that can create the sender successfully. Yet it would break if we re-deployed it with irrelevant changes.

@Zhiyi Li thanks for your patienceon this, it seems like a very difficult problem to debug! You said you used the same sarma client version to connect (presumably outside of Flyte) with no issue, is this correct? Are there any configuration options in that connection that we are not setting in Flyte?

We found the problem was the json decoding of the kafka version did not work. Can someone takes a further look to see how to fix it?

I'm not sure I fully understand, can you elaborate on this? Is this an issue in a Flyte repo or sarma client?

This is a flyte repo problem I suspect. The input interface of the kafkaconfig inside cloud_event reads in as

version: 
  verion: []int

Yet the admin accessor expects the config to be

Version:
  version:[]int

Some part of the above struct conversion is not working properly and the decode fails…

Ah, OK. I think I'm understanding. When decoding the KafkaConfig FlyteAdmin expects a capitalized "Version". In the screenshot you provide the config you specified is

[2,2,0,0]

and it was incorrectly parsed as

[0,0,0,0]

. Is this correct? Have you tried updating your config to:

cloudevents:
    enable: true
    kafka:
      brokers: kafkaip
      Version:
        version:
          - 2
          - 2
          - 0
          - 0
    eventsPublisher:
      eventTypes:
      - all
      topicName: workflow-engine-test
    type: Kafka

I'm wondering if there is a specific reason we didn't annotate the

Version

with a JSON name because it's the only way to properly decode the

sarama.KafkaVersion

.

That sounds reasonable to me

cc @Kevin Su ^^^

Hi @Zhiyi Li. I just created a PR to fix it. Mind taking a look?