Strimzi Kafka Bridge(桥接)实战之三:自制sdk(golang版本)

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos

本篇概览

  • 本文是《Strimzi Kafka Bridge(桥接)实战》的第三篇,前文咱们掌握了Strimzi Kafka Bridge的基本功能:基于http提供各种kafka消息的服务
  • 此刻,如果想通过http接口调用bridge的服务,势必要写不少代码(请求数据的生成、响应数据的解析),好在Strimzi已经提供了标准OpenApi的配置文件,咱们可以根据这个配置文件生成与http接口相关的代码,省去不少工作

为什么是golang版本

  • 熟悉欣宸的读者都知道欣宸是个正宗的java程序员,那么,本篇应该实战java版本的SDK吧,怎么就研究起了golang版本呢?
  • 因为Strimzi Kafka Bridge提供的OpenApi配置,用来生成客户端sdk之后,是无法正常使用的!!!,没错,您没看错,用工具生成的sdk,不论是golang版还是java版,都用不了!
  • 相比之下,golang版的sdk,虽然不能用,但是经过抢救还是可以正常工作的,这也是本篇的主要内容
  • 而java版的就没那么幸运了,涉及到jar库的依赖,就算是改代码也救不活,于是只能放弃,具体的原因本文末尾会给出,当然了,也许是欣宸水平太差,换成其他高手说不定就给救活了
  • 闲话少说,接下来的内容由以下这几个步骤组成
  1. 介绍一下我这边的环境信息
  2. 下载OpenApi的配置文件
  3. 下载swagger工具
  4. 用swagger工具生成客户端sdk代码
  5. 创建一个golang的demo程序,使用刚刚生成的客户端sdk代码
  6. 客户端sdk代码存在诸多问题,但是可以逐个修复,这里咱们就来修复它们
  7. 运行一个demo程序,调用sdk代码中的API,验证基本功能

环境信息

  • 以下是我这边的环境信息,您可以作为参考
  1. JDK:11.0.14.1
  2. Maven:3.8.5
  3. strimzi-kafka-bridge:0.22.3
  4. swagger-codegen-cli:2.4.9
  • 需要注意的是,swagger工具是jar格式的,因此需要当前环境准备好JDK

下载OpenApi的配置文件

  • Strimzi Kafka Bridge的master分支处于活跃状态,因此不适合拿来实战,咱们选择一个发布版本吧
  • 下载strimzi-kafka-bridge源码,地址是:https://codeload.github.com/strimzi/strimzi-kafka-bridge/zip/refs/tags/0.22.3 ,下载后解压得到名为strimzi-kafka-bridge-0.22.3的文件夹
  • 这个文件就是OpenApi的配置文件,可以用来生成客户端sdk源码:strimzi-kafka-bridge-0.22.3/src/main/resources/openapiv2.json ,稍后会用到

下载swagger工具

用swagger工具生成客户端sdk代码

  • 使用默认参数来生成客户端sdk代码的操作十分简单
java -jar swagger-codegen-cli-2.4.9.jar generate \
-i ./openapiv2.json \
-l go \
-o swagger
  • 执行完命令后,控制台输出如下

  • 查看swagger目录,发现已经生成了大量文件

➜  001 tree swagger
swagger
├── README.md
├── api
│   └── swagger.yaml
├── api_consumers.go
├── api_default.go
├── api_producer.go
├── api_seek.go
├── api_topics.go
├── client.go
├── configuration.go
├── docs
│   ├── AssignedTopicPartitions.md
│   ├── BridgeInfo.md
│   ├── Consumer.md
│   ├── ConsumerRecord.md
│   ├── ConsumerRecordList.md
│   ├── ConsumersApi.md
│   ├── CreatedConsumer.md
│   ├── DefaultApi.md
│   ├── KafkaHeader.md
│   ├── KafkaHeaderList.md
│   ├── ModelError.md
│   ├── OffsetCommitSeek.md
│   ├── OffsetCommitSeekList.md
│   ├── OffsetRecordSent.md
│   ├── OffsetRecordSentList.md
│   ├── OffsetsSummary.md
│   ├── Partition.md
│   ├── PartitionMetadata.md
│   ├── Partitions.md
│   ├── ProducerApi.md
│   ├── ProducerRecord.md
│   ├── ProducerRecordList.md
│   ├── ProducerRecordToPartition.md
│   ├── ProducerRecordToPartitionList.md
│   ├── Replica.md
│   ├── SeekApi.md
│   ├── SubscribedTopicList.md
│   ├── TopicMetadata.md
│   ├── Topics.md
│   └── TopicsApi.md
├── git_push.sh
├── model_assigned_topic_partitions.go
├── model_bridge_info.go
├── model_consumer.go
├── model_consumer_record.go
├── model_consumer_record_list.go
├── model_created_consumer.go
├── model_error.go
├── model_kafka_header.go
├── model_kafka_header_list.go
├── model_offset_commit_seek.go
├── model_offset_commit_seek_list.go
├── model_offset_record_sent.go
├── model_offset_record_sent_list.go
├── model_offsets_summary.go
├── model_partition.go
├── model_partition_metadata.go
├── model_partitions.go
├── model_producer_record.go
├── model_producer_record_list.go
├── model_producer_record_to_partition.go
├── model_producer_record_to_partition_list.go
├── model_replica.go
├── model_subscribed_topic_list.go
├── model_topic_metadata.go
├── model_topics.go
└── response.go

2 directories, 66 files

创建一个golang的demo程序,使用刚刚生成的客户端sdk代码

  • 新建名为sdkdemo的文件夹
  • sdkdemo的文件夹下面执行以下命令,新建一个go工程
go mod init sdkdemo
  • 需要引入两个包,执行以下命令
go get golang.org/x/oauth2
go get github.com/antihax/optional
  • 将前面生成代码的swagger文件夹复制到sdkdemo的文件夹下面

  • 现在sdkdemo的文件夹下面有这些东西

  • 为了方便开发,接下来用IDE工具进行开发,我这里用的是goland,打开项目后新增名为main.go的文件

  • 接下来咱们要面对的是一堆破绽百出的sdk代码,不过还好,可以拯救,咱们一起啦拯救吧

修复有问题的sdk源码,第一个问题

  • 一共有6个问题,咱们逐一修复
  • 第一个问题如下图,SeekToEndOpts这个数据结构在api_seek.goapi_consumer.go中都有,显然是重复定义了,将左侧api_seek.go中的SeekToEndOpts定义删除掉

第二个问题

  • 第二个问题如下图,SendOpts这个数据结构在api_topics.goapi_producer.go中都有,显然是重复定义了,将左侧api_topics.go中的SeekToEndOpts定义删除掉

第三个问题

  • 第三个问题最让人痛苦(因为java版也被此问题折磨,且不好处理),bridge的请求和响应的contentType,与咱们平时常用的application/json不同,在bridge这里用的是这两种:application/vnd.kafka.v2+jsonapplication/vnd.kafka.json.v2+json,其实这个也好理解:生产和发送的消息内容不一定只有json格式,可能还会嵌入其他格式的消息,这就要有kafka自己的协议来支持了,于是contentType就变得比较特殊
  • 话虽这么说,但是swagger不认识application/vnd.kafka.v2+jsonapplication/vnd.kafka.json.v2+json这两种格式,于是生成的代码自然也就不支持了
  • 来看看具体问题吧,打开文件client.go,当前decode方法源码如下,可见是不会处理application/vnd.kafka.v2+jsonapplication/vnd.kafka.json.v2+json这两种的
func (c *APIClient) decode(v interface{}, b []byte, contentType string) (err error) {
	if strings.Contains(contentType, "application/xml") {
		if err = xml.Unmarshal(b, v); err != nil {
			return err
		}
		return nil
	} else if strings.Contains(contentType, "application/json") {
		if err = json.Unmarshal(b, v); err != nil {
			return err
		}
		return nil
	}
	return errors.New("undefined response type")
}
  • 把代码改成下面这样,对application/vnd.kafka.v2+jsonapplication/vnd.kafka.json.v2+json这两种类型的数据,处理方法都等同于json
func (c *APIClient) decode(v interface{}, b []byte, contentType string) (err error) {
	if strings.Contains(contentType, "application/xml") {
		if err = xml.Unmarshal(b, v); err != nil {
			return err
		}
		return nil
	} else if strings.Contains(contentType, "application/json") ||
		strings.Contains(contentType, "application/vnd.kafka.v2+json") ||
		strings.Contains(contentType, "application/vnd.kafka.json.v2+json") {
		if err = json.Unmarshal(b, v); err != nil {
			return err
		}
		return nil
	}
	return errors.New("undefined response type")
}
  • 当然了这样做的弊端也很明显:只支持json格式的内容,kakfa原本支持的多种格式都不能处理了

第四个问题

  • 第四个问题也和contentType有关,前面第三个问题发生在请求阶段,而第四个问题发生在处理响应数据的阶段
  • 还是client.go文件,这次是setBody方法,先看看原始内容
// Set request body from an interface{}
func setBody(body interface{}, contentType string) (bodyBuf *bytes.Buffer, err error) {
	if bodyBuf == nil {
		bodyBuf = &bytes.Buffer{}
	}

	if reader, ok := body.(io.Reader); ok {
		_, err = bodyBuf.ReadFrom(reader)
	} else if b, ok := body.([]byte); ok {
		_, err = bodyBuf.Write(b)
	} else if s, ok := body.(string); ok {
		_, err = bodyBuf.WriteString(s)
	} else if s, ok := body.(*string); ok {
		_, err = bodyBuf.WriteString(*s)
	} else if jsonCheck.MatchString(contentType) {
		err = json.NewEncoder(bodyBuf).Encode(body)
	} else if xmlCheck.MatchString(contentType) {
		xml.NewEncoder(bodyBuf).Encode(body)
	}

	if err != nil {
		return nil, err
	}

	if bodyBuf.Len() == 0 {
		err = fmt.Errorf("Invalid body type %s\n", contentType)
		return nil, err
	}
	return bodyBuf, nil
}
  • 修改后的内容如下图,红色箭头所指为新增内容

第五个问题

  • 第五个问题,简直是strimzi拿来恶心开发者的,在拉取消息的时候,bridge的server端只支持application/vnd.kafka.json.v2+json,结果在OpenApi中却定义了多种类型,结果拉去消息的时候,bridge会提示多出的类型不支持
  • 这个问题可以用postman等工具复现,如下图
  • 代码的改动如下图,修改api_consumers.go

第六个问题

  • 最后一个问题是数据结构定义问题,打开model_consumer_record_list.go,看到内容如下,真够坏的,挖这么大的坑...
package swagger

type ConsumerRecordList struct {
}
  • 改成这样就好了
package swagger

type ConsumerRecordList []ConsumerRecord

第七个问题

  • 第七个问题,也是挖了个坑让我跳,打开文件model_producer_record.go,内容如下,根据前一篇的请求内容,可知这里缺少两个字段:KeyValue
package swagger

type ProducerRecord struct {
	Partition int32 `json:"partition,omitempty"`
	Headers *KafkaHeaderList `json:"headers,omitempty"`
}
  • 修改后如下
package swagger

type ProducerRecord struct {
	Partition int32 `json:"partition,omitempty"`
	Value string `json:"value"`
	Key string `json:"key,omitempty"`
	Headers *KafkaHeaderList `json:"headers,omitempty"`
}

第八个问题

  • 最后一个问题,是在提交offset的时候,bridge后台不接受contentType,所以请打开文件api_consumers.go,修改如下,注释掉一行代码

  • 坑已经填完了,开始验证SDK能不能用吧

编写代码验证功能:查看topic列表

  • 打开main.go文件,增加以下内容,都是要用到的常量,以及sdk配置的初始化
// 测试用的topic
const TEST_TOPIC = "bridge-quickstart-topic"

const TEST_GROUP = "client-sdk-group"

const CONSUMER_NAME = "client-sdk-consumer-002"

// strimzi bridge地址
const BASE_PATH = "http://127.0.0.1:31331"

var client *swagger.APIClient

func init() {
	configuration := swagger.NewConfiguration()
	configuration.BasePath = BASE_PATH
	client = swagger.NewAPIClient(configuration)
}
  • 调用SDK来查看kafka的topic列表的代码如下
func getAllTopics() ([]string, error) {
	array, response, err := client.TopicsApi.ListTopics(context.Background())

	if err != nil {
		log.Printf("getAllTopics err: %v\n", err)
		return nil, err
	}

	log.Printf("response: %v", response)

	return array, nil
}
  • 在main方法中调用getAllTopics
func main() {
	topics, err := getAllTopics()
	if err != nil {
		return
	}

	fmt.Printf("topics: %v\n", topics)
}
  • 运行main方法,结果如下,可见成功获取到topic列表,sdk能用
2022/12/18 21:26:33 response: &{200 OK 200 HTTP/1.1 1 1 map[Content-Length:[109] Content-Type:[application/vnd.kafka.v2+json]] 0x140000e0300 109 [] false false map[] 0x14000118100 <nil>}
topics: [__strimzi_store_topic bridge-quickstart-topic __strimzi-topic-operator-kstreams-topic-store-changelog]

Process finished with the exit code 0

编写代码验证功能:发送消息

  • 发送消息的代码如下
// 发送消息(异步模式,不会收到offset返回)
func sendAsync(info string) error {
	log.Print("send [" + info + "]")
	_, response, err := client.ProducerApi.Send(context.Background(),
		TEST_TOPIC,
		swagger.ProducerRecordList{
			Records: []swagger.ProducerRecord{
				{Value: "message from go swagger SDK"},
			},
		},
		&swagger.SendOpts{Async: optional.NewBool(true)},
	)

	if err != nil {
		log.Printf("send err: %v\n", err)
		return err
	}

	log.Printf("response: %v", response.StatusCode)

	return nil
}
  • 把main方法改成下面这样,连续调用发送消息的请求
func main() {
	for i := 0; i < 10; i++ {
		sendAsync("message from go client " + strconv.Itoa(i))
	}
}
  • 控制台输出如下,可见发送消息成功,稍后咱们还会写消费的代码来消费这些消息
/private/var/folders/5v/p3bj9bzx2nd99y5l21nb1c080000gn/T/GoLand/___go_build_sdkdemo
2022/12/18 21:35:47 send [message from go client 0]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 1]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 2]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 3]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 4]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 5]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 6]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 7]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 8]
2022/12/18 21:35:47 response: 204
2022/12/18 21:35:47 send [message from go client 9]
2022/12/18 21:35:47 response: 204

Process finished with the exit code 0

编写代码验证功能:创建consumer

  • 先增加两个辅助方法,用来处理特别的包体和错误信息
// 取出swagger特有的error类型,从中提取中有效的错误信息
func getErrorMessage(err error) string {
	e := err.(swagger.GenericSwaggerError)
	return string(e.Body())
}

func getBodyStr(body io.ReadCloser) string {
	buf := new(bytes.Buffer)
	buf.ReadFrom(body)
	return buf.String()
}

  • 创建consumer的代码如下
// 创建consumer
func CreateConsumer(group string, consumerName string) (*swagger.CreatedConsumer, error) {

	consumer, response, err := client.ConsumersApi.CreateConsumer(context.Background(),
		group,
		swagger.Consumer{
			Name:                     consumerName,
			AutoOffsetReset:          "latest",
			FetchMinBytes:            16,
			ConsumerRequestTimeoutMs: 300 * 1000,
			EnableAutoCommit:         false,
			Format:                   "json",
		})

	if err != nil {
		log.Printf("CreateConsumer error : %v", getErrorMessage(err))
		return nil, err
	}

	log.Printf("CreateConsumer response : %v, body [%v]", response, getBodyStr(response.Body))
	log.Printf("consumer : %v", consumer)
	return &consumer, nil
}
  • 在main方法中调用,即可创建consumer
func main() {
	// 创建consumer
	CreateConsumer(TEST_GROUP, CONSUMER_NAME)
}

编写代码验证功能:订阅

  • 订阅代码如下
// 订阅
func Subsciribe(topic string, consumerGroup string, consumerName string) error {

	response, err := client.ConsumersApi.Subscribe(context.Background(),
		swagger.Topics{Topics: []string{topic}},
		consumerGroup,
		consumerName,
	)

	if err != nil {
		log.Printf("Subscribe error : %v", err)
		return err
	}

	log.Printf("Subscribe response : %v", response)
	return nil
}
  • 在main方法中这样调用
func main() {
	err := Subsciribe(TEST_TOPIC, TEST_GROUP, CONSUMER_NAME)
	if err != nil {
		fmt.Printf("err : %v\n", err)
	}
}

编写代码验证功能:拉取消息

  • 以下是拉取消息的代码
// 拉取消息
func Poll(consumerGroup string, consumerName string) error {
	// ctx context.Context, groupid string, name string, localVarOptionals *PollOpts
	recordList, response, err := client.ConsumersApi.Poll(context.Background(), consumerGroup, consumerName, nil)
	if err != nil {
		log.Printf("Poll error : %v", err)
		return err
	}

	log.Printf("Poll response : %v", response)
	fmt.Printf("recordList: %v\n", recordList)
	return nil
}
  • main方法如下
func main() {
	Poll(TEST_GROUP, CONSUMER_NAME)
}
  • 执行main方法,第一次拉取不到消息,别担心,这是正常的现象,按照官方的说法,拉取到的第一条消息就是空的,这是因为拉取操作出触发了rebalancing逻辑(rebalancing是kafka的概览,是处理多个partition消费的操作),再次执行main方法,这下正常了,控制台输出如下
/private/var/folders/5v/p3bj9bzx2nd99y5l21nb1c080000gn/T/GoLand/___go_build_sdkdemo
2022/12/18 21:43:16 Poll response : &{200 OK 200 HTTP/1.1 1 1 map[Content-Length:[2301] Content-Type:[application/vnd.kafka.json.v2+json]] 0x140000e0340 2301 [] false false map[] 0x1400011a100 <nil>}
recordList: [{ 163468 0 bridge-quickstart-topic message from go swagger SDK <nil>} { 163469 0 bridge-quickstart-topic message from go swagger SDK <nil>} { 163470 0 bridge-quickstart-topic message from go swagger SDK <nil>} { 163471 0 bridge-quickstart-topic message from go swagger SDK <nil>} { 163472 0 bridge-quickstart-topic message from go swagger SDK <nil>} { 163473 0 bridge-quickstart-topic message from go swagger SDK <nil>} { 162246 2 bridge-quickstart-topic message from go swagger SDK <nil>} { 162247 2 bridge-quickstart-topic message from go swagger SDK <nil>} { 162248 2 bridge-quickstart-topic message from go swagger SDK <nil>} { 162249 2 bridge-quickstart-topic message from go swagger SDK <nil>} { 162250 2 bridge-quickstart-topic message from go swagger SDK <nil>} { 163669 1 bridge-quickstart-topic message from go swagger SDK <nil>} { 163670 1 bridge-quickstart-topic message from go swagger SDK <nil>} { 163671 1 bridge-quickstart-topic message from go swagger SDK <nil>} { 163672 1 bridge-quickstart-topic message from go swagger SDK <nil>} { 163146 3 bridge-quickstart-topic message from go swagger SDK <nil>} { 163147 3 bridge-quickstart-topic message from go swagger SDK <nil>} { 163148 3 bridge-quickstart-topic message from go swagger SDK <nil>} { 163149 3 bridge-quickstart-topic message from go swagger SDK <nil>} { 163150 3 bridge-quickstart-topic message from go swagger SDK <nil>}]

Process finished with the exit code 0

编写代码验证功能:提交offset

  • 最后是提交offset的功能,这样从消息的发送再到接收的整个流程都实现了api覆盖,增加Offset方法
// 提交offset
func Offset(consumerGroup string, consumerName string) error {
	response, err := client.ConsumersApi.Commit(context.Background(),
		consumerGroup,
		consumerName, nil)

	if err != nil {
		log.Printf("Poll error : %v", err)
		return err
	}

	log.Printf("Offset response : %v", response)
	return nil
}

  • 调用很简单
func main() {
	err := Offset(TEST_GROUP, CONSUMER_NAME)
	if err != nil {
		print(err)
	}
}

  • 执行结果如下,返回204,提交成功
/private/var/folders/5v/p3bj9bzx2nd99y5l21nb1c080000gn/T/GoLand/___go_build_sdkdemo
2022/12/18 22:07:38 Offset response : &{204 No Content 204 HTTP/1.1 1 1 map[] {} 0 [] false false map[] 0x1400011a100 <nil>}

Process finished with the exit code 0

java的问题

  • 从go版本的修改程度可以发现,基于openapiv2.json生成的sdk代码真的很难用,在go环境尚且如此,换成java环境就更难改了,虽然我也尝试过将其改好,但是面对很多jar的时候还是无能为力,下图是一个很难处理的地方,ApiClient并不支持application/vnd.kafka.v2+jsonapplication/vnd.kafka.json.v2+json,contentType改不成正常的,bridge后台就会返回错误,所以最终我只能骂骂咧咧的放弃了

有收获吗?

  • 面对这么烂的SDK源码,一般人都不会在生产环境使用,但是个人觉得也不是一无是处,这里小结一下收获
  1. 了解了go版本swagger sdk源码的基本结构,和请求响应逻辑
  2. 知道了大众工具也有出问题的时候
  3. strimzi到底测试过吗,这个做CICD自动化应该可以做到吧,能进CNCF的项目,也是会出问题的...

欢迎关注博客园:程序员欣宸

学习路上,你不孤单,欣宸原创一路相伴...

热门相关:总裁的秘密爱人   网游三国之城市攻略   网游三国之城市攻略   嫡嫁千金   重生之嫡女祸妃