diff --git a/integrations/README.md b/integrations/README.md new file mode 100644 index 0000000000..217700aae9 --- /dev/null +++ b/integrations/README.md @@ -0,0 +1,27 @@ +# Integration tests for KoP + +# Produce and Consume support + +## Golang + +* [https://github.com/Shopify/sarama](https://github.com/Shopify/sarama) +* [https://github.com/confluentinc/confluent-kafka-go](https://github.com/confluentinc/confluent-kafka-go) + +## Rust + +* [https://github.com/fede1024/rust-rdkafka](https://github.com/fede1024/rust-rdkafka) + +## NodeJS + +* [https://github.com/Blizzard/node-rdkafka](https://github.com/Blizzard/node-rdkafka) + +# Partial support + +### [kafka-node](https://www.npmjs.com/package/kafka-node) + +Producing is working, but consuming is failing as the library is sending FETCH v0 regardless of API_VERSIONS responses: + +``` +DEBUG io.streamnative.kop.KafkaCommandDecoder - Write kafka cmd response back to client. request: RequestHeader(apiKey=FETCH, apiVersion=0, clientId=kafka-node-client, correlationId=4) +INFO io.streamnative.kop.KafkaIntegrationTest - STDOUT: Error: Not a message set. Magic byte is 2 +``` \ No newline at end of file diff --git a/integrations/golang-confluent-kafka/Dockerfile b/integrations/golang-confluent-kafka/Dockerfile new file mode 100644 index 0000000000..08e203df84 --- /dev/null +++ b/integrations/golang-confluent-kafka/Dockerfile @@ -0,0 +1,39 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +FROM golang:1.13.4-alpine + +RUN apk -U add ca-certificates git bash build-base sudo pkgconf build-base +RUN git clone https://github.com/edenhill/librdkafka.git && cd librdkafka && ./configure --prefix /usr && make && make install + +WORKDIR /go/src/app +COPY . . + +RUN go get -d -v ./... +RUN go install -v ./... + +CMD sh -c '/go/bin/golang-confluent-kafka; echo "ExitCode=$?"' \ No newline at end of file diff --git a/integrations/golang-confluent-kafka/go.mod b/integrations/golang-confluent-kafka/go.mod new file mode 100644 index 0000000000..21f587579c --- /dev/null +++ b/integrations/golang-confluent-kafka/go.mod @@ -0,0 +1,5 @@ +module github.com/apache/pulsar/kop/integration/golang-confluent-kafka + +go 1.13 + +require gopkg.in/confluentinc/confluent-kafka-go.v1 v1.1.0 diff --git a/integrations/golang-confluent-kafka/go.sum b/integrations/golang-confluent-kafka/go.sum new file mode 100644 index 0000000000..21841b7e75 --- /dev/null +++ b/integrations/golang-confluent-kafka/go.sum @@ -0,0 +1,2 @@ +gopkg.in/confluentinc/confluent-kafka-go.v1 v1.1.0 h1:roy97m/3wj9/o8OuU3sZ5wildk30ep38k2x8nhNbKrI= +gopkg.in/confluentinc/confluent-kafka-go.v1 v1.1.0/go.mod h1:ZdI3yfYmdNSLQPNCpO1y00EHyWaHG5EnQEyL/ntAegY= diff --git a/integrations/golang-confluent-kafka/main.go b/integrations/golang-confluent-kafka/main.go new file mode 100644 index 0000000000..e3941f4668 --- /dev/null +++ b/integrations/golang-confluent-kafka/main.go @@ -0,0 +1,141 @@ +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package main + +import ( + "fmt" + "os" + "strconv" + "strings" + + "gopkg.in/confluentinc/confluent-kafka-go.v1/kafka" +) + +func main() { + limit, err := strconv.Atoi(getEnv("KOP_LIMIT", "10")) + if err != nil { + panic(err) + } + + shouldProduce, err := strconv.ParseBool(getEnv("KOP_PRODUCE", "false")) + if err != nil { + panic(err) + } + + shouldConsume, err := strconv.ParseBool(getEnv("KOP_CONSUME", "false")) + if err != nil { + panic(err) + } + + brokers := []string{getEnv("KOP_BROKER", "localhost:9092")} + topic := getEnv("KOP_TOPIC", "my-confluent-go-topic") + topics := []string{topic} + + if shouldProduce { + + fmt.Println("starting to produce") + + p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": getEnv("KOP_BROKER", "localhost:9092")}) + if err != nil { + panic(err) + } + defer p.Close() + + // Delivery report handler for produced messages + go func() { + for e := range p.Events() { + switch ev := e.(type) { + case *kafka.Message: + if ev.TopicPartition.Error != nil { + fmt.Printf("Delivery failed: %v\n", ev.TopicPartition) + } else { + fmt.Printf("Delivered message to %v\n", ev.TopicPartition) + } + } + } + }() + + for i := 0; i < limit; i++ { + p.Produce(&kafka.Message{ + TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, + Value: []byte("hello from confluent go"), + }, nil) + fmt.Println("send a message") + + } + fmt.Printf("produced all messages successfully (%d) \n", limit) + // Wait for message deliveries before shutting down + p.Flush(15 * 1000) + + } + + if shouldConsume { + fmt.Println("starting to consume") + + c, err := kafka.NewConsumer(&kafka.ConfigMap{ + "bootstrap.servers": strings.Join(brokers, ","), + "group.id": "myGroup", + "auto.offset.reset": "earliest", + "broker.version.fallback": "2.0.0", + "debug": "all", + }) + if err != nil { + panic(err) + } + + c.SubscribeTopics(topics, nil) + + counter := 0 + + for counter < limit { + msg, err := c.ReadMessage(-1) + if err == nil { + fmt.Println("received msg") + fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value)) + counter++ + } else { + // The client will automatically try to recover from all errors. + fmt.Printf("Consumer error: %v (%v)\n", err, msg) + panic(err) + } + } + fmt.Println("consumed all messages successfully") + } + + fmt.Println("exiting normally") +} + +func getEnv(key, fallback string) string { + value, exists := os.LookupEnv(key) + if !exists { + value = fallback + } + return value +} diff --git a/integrations/golang-sarama/Dockerfile b/integrations/golang-sarama/Dockerfile new file mode 100644 index 0000000000..4a10bc2280 --- /dev/null +++ b/integrations/golang-sarama/Dockerfile @@ -0,0 +1,36 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +FROM golang:1.13.4-stretch + +WORKDIR /go/src/app +COPY . . + +RUN go get -d -v ./... +RUN go install -v ./... + +CMD sh -c '/go/bin/sarama-golang; echo "ExitCode=$?"' \ No newline at end of file diff --git a/integrations/golang-sarama/go.mod b/integrations/golang-sarama/go.mod new file mode 100644 index 0000000000..c13ce190b7 --- /dev/null +++ b/integrations/golang-sarama/go.mod @@ -0,0 +1,5 @@ +module github.com/apache/pulsar/kop/integration/sarama-golang + +go 1.13 + +require github.com/Shopify/sarama v1.24.1 diff --git a/integrations/golang-sarama/go.sum b/integrations/golang-sarama/go.sum new file mode 100644 index 0000000000..f327335702 --- /dev/null +++ b/integrations/golang-sarama/go.sum @@ -0,0 +1,52 @@ +github.com/Shopify/sarama v1.24.1 h1:svn9vfN3R1Hz21WR2Gj0VW9ehaDGkiOS+VqlIcZOkMI= +github.com/Shopify/sarama v1.24.1/go.mod h1:fGP8eQ6PugKEI0iUETYYtnP6d1pH/bdDMTel1X5ajsU= +github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/eapache/go-resiliency v1.1.0 h1:1NtRmCAqadE2FN4ZcN6g90TP3uk8cg9rn9eNK2197aU= +github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs= +github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw= +github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU= +github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= +github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= +github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= +github.com/frankban/quicktest v1.4.1/go.mod h1:36zfPVQyHxymz4cH7wlDmVwDrJuljRB60qkgn7rorfQ= +github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/hashicorp/go-uuid v1.0.1 h1:fv1ep09latC32wFoVwnqcnKJGnMSdBanPczbHAYm1BE= +github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= +github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjLBdCp5PRlCFijNjvcYANOZXzCfXwCM= +github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o= +github.com/klauspost/compress v1.8.2 h1:Bx0qjetmNjdFXASH02NSAREKpiaDwkO1DRZ3dV2KCcs= +github.com/klauspost/compress v1.8.2/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/pierrec/lz4 v2.2.6+incompatible h1:6aCX4/YZ9v8q69hTyiR7dNLnTA3fgtKHVVW5BCd5Znw= +github.com/pierrec/lz4 v2.2.6+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ= +github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I= +github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5 h1:bselrhR0Or1vomJZC8ZIjWtbDmn9OYFLX5Ik9alpJpE= +golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3 h1:0GoQqolDA55aaLxZyTzK/Y2ePZzZTUrRacwib7cNsYQ= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw= +gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM= +gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q= +gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4= +gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc44010= +gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM= +gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU= +gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8= diff --git a/integrations/golang-sarama/main.go b/integrations/golang-sarama/main.go new file mode 100644 index 0000000000..ce24337472 --- /dev/null +++ b/integrations/golang-sarama/main.go @@ -0,0 +1,164 @@ +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// +package main + +import ( + "context" + "fmt" + "log" + "os" + "strconv" + "sync" + + "github.com/Shopify/sarama" +) + +type exampleConsumerGroupHandler struct { + counter int + limit int + wg *sync.WaitGroup +} + +func (exampleConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil } +func (exampleConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil } +func (h exampleConsumerGroupHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { + for msg := range claim.Messages() { + fmt.Printf("Message topic:%q partition:%d offset:%d\n", msg.Topic, msg.Partition, msg.Offset) + sess.MarkMessage(msg, "") + h.counter++ + fmt.Printf("received msg %d/%d\n", h.counter, h.limit) + if h.counter == h.limit { + fmt.Println("consumed all messages successfully") + h.wg.Done() + return nil + } + } + return nil +} + +func main() { + + limit, err := strconv.Atoi(getEnv("KOP_LIMIT", "10")) + if err != nil { + panic(err) + } + + shouldProduce, err := strconv.ParseBool(getEnv("KOP_PRODUCE", "false")) + if err != nil { + panic(err) + } + + shouldConsume, err := strconv.ParseBool(getEnv("KOP_CONSUME", "false")) + if err != nil { + panic(err) + } + + // Init config, specify appropriate version + config := sarama.NewConfig() + config.Version = sarama.V2_0_0_0 + config.Metadata.Retry.Max = 0 + config.Consumer.Return.Errors = true + config.Consumer.Offsets.Initial = sarama.OffsetOldest + config.Producer.Return.Successes = true + brokers := []string{getEnv("KOP_BROKER", "localhost:9092")} + topic := getEnv("KOP_TOPIC", "my-sarama-topic") + topics := []string{topic} + sarama.Logger = log.New(os.Stdout, "", log.Ltime) + + fmt.Println("connecting to", brokers) + + // Start with a client + client, err := sarama.NewClient(brokers, config) + if err != nil { + panic(err) + } + defer func() { _ = client.Close() }() + + var waitgroup sync.WaitGroup + if shouldConsume { + waitgroup.Add(1) + + // Start a new consumer group + group, err := sarama.NewConsumerGroupFromClient("sarama-consumer", client) + if err != nil { + panic(err) + } + defer func() { _ = group.Close() }() + + fmt.Println("starting to consume") + + // Iterate over consumer sessions. + ctx := context.Background() + handler := exampleConsumerGroupHandler{counter: 0, limit: limit, wg: &waitgroup} + + err = group.Consume(ctx, topics, handler) + if err != nil { + panic(err) + } + + } + + if shouldProduce { + syncProducer, err := sarama.NewSyncProducerFromClient(client) + if err != nil { + panic(err) + } + defer func() { _ = syncProducer.Close() }() + + fmt.Println("starting to produce") + + for i := 0; i < limit; i++ { + msg := &sarama.ProducerMessage{ + Topic: topic, + Value: sarama.StringEncoder("hello from sarama"), + Metadata: "test", + } + + fmt.Println("send a message") + + _, _, err := syncProducer.SendMessage(msg) + if err != nil { + panic(err) + } + } + fmt.Printf("produced all messages successfully (%d) \n", limit) + + } + waitgroup.Wait() + fmt.Println("exiting normally") + +} + +func getEnv(key, fallback string) string { + value, exists := os.LookupEnv(key) + if !exists { + value = fallback + } + return value +} diff --git a/integrations/node-kafka-node/Dockerfile b/integrations/node-kafka-node/Dockerfile new file mode 100644 index 0000000000..11092a9727 --- /dev/null +++ b/integrations/node-kafka-node/Dockerfile @@ -0,0 +1,39 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +FROM node:13-alpine + +RUN apk add yarn snappy + +# Create app directory +WORKDIR /usr/src/app + +COPY . . + +RUN yarn install + +CMD sh -c 'node main.js; echo "ExitCode=$?"' \ No newline at end of file diff --git a/integrations/node-kafka-node/main.js b/integrations/node-kafka-node/main.js new file mode 100644 index 0000000000..0ab0e411f6 --- /dev/null +++ b/integrations/node-kafka-node/main.js @@ -0,0 +1,80 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +'use strict'; + +var kafka = require('kafka-node'); +var HighLevelProducer = kafka.HighLevelProducer; +var Consumer = kafka.Consumer; +var broker = process.env.KOP_BROKER || 'localhost:9092'; +var topic = process.env.KOP_TOPIC || 'topic1'; +var limit = parseInt(process.env.KOP_LIMIT || 10, 10); +var should_produce = process.env.KOP_PRODUCE || false; +var should_consume = process.env.KOP_CONSUME || false; + +var counter = 0; + +var Client = kafka.KafkaClient; +var client = new Client({ kafkaHost: broker }); + +if (should_consume) { + // if fetchMaxBytes is set too low the broker could start sending fetch responses in RecordBatch format instead of MessageSet :( + // https://www.npmjs.com/package/kafka-node#error-not-a-message-set-magic-byte-is-2 + var consumer = new Consumer(client, [{ topic: topic, partition: 0, fetchMaxBytes: 1024 * 1024 * 1024 * 1024 }], { + autoCommit: true + }); + console.log('starting to consume'); + + consumer.on("message", function (message) { + console.log('received msg ' + message); + counter++; + if (counter == limit) { + console.log("consumed all messages successfully"); + process.exit(); + }; + }); + + consumer.on("error", function (err) { + console.log(err); + process.exit(1); + }); +} + +if (should_produce) { + var producer = new HighLevelProducer(client); + producer.on('ready', function () { + console.log('starting to produce'); + setInterval(send, 500); + }); + + producer.on('error', function (err) { + console.log('error', err); + process.exit(1); + }); +} + +function send() { + var message = new Date().toString(); + producer.send([{ topic: topic, messages: [message] }], function (err, data) { + if (err) { + console.log('error', err); + process.exit(1); + } + counter++; + if (counter == limit) { + console.log("produced all messages successfully"); + process.exit(); + }; + }); +} \ No newline at end of file diff --git a/integrations/node-kafka-node/package.json b/integrations/node-kafka-node/package.json new file mode 100644 index 0000000000..59c0358b30 --- /dev/null +++ b/integrations/node-kafka-node/package.json @@ -0,0 +1,9 @@ +{ + "name": "node-kafka-node", + "version": "1.0.0", + "main": "index.js", + "license": "Apache-2.0", + "dependencies": { + "kafka-node": "^5.0.0" + } +} diff --git a/integrations/node-rdkafka/Dockerfile b/integrations/node-rdkafka/Dockerfile new file mode 100644 index 0000000000..7187ebaf08 --- /dev/null +++ b/integrations/node-rdkafka/Dockerfile @@ -0,0 +1,53 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +FROM node:8-alpine + +RUN apk add yarn snappy + +RUN apk --no-cache add \ + bash \ + g++ \ + ca-certificates \ + lz4-dev \ + musl-dev \ + cyrus-sasl-dev \ + openssl-dev \ + make \ + python + +RUN apk add --no-cache --virtual .build-deps gcc zlib-dev libc-dev bsd-compat-headers py-setuptools bash + + +# Create app directory +WORKDIR /usr/src/app + +COPY . . + +RUN yarn install + +CMD sh -c 'node main.js; echo "ExitCode=$?"' \ No newline at end of file diff --git a/integrations/node-rdkafka/main.js b/integrations/node-rdkafka/main.js new file mode 100644 index 0000000000..42f7292dcb --- /dev/null +++ b/integrations/node-rdkafka/main.js @@ -0,0 +1,161 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// +'use strict'; +const Kafka = require('node-rdkafka'); + +var broker = process.env.KOP_BROKER || 'localhost:9092'; +var topic = process.env.KOP_TOPIC || 'topic1'; +var limit = parseInt(process.env.KOP_LIMIT || 10, 10); +var should_produce = process.env.KOP_PRODUCE || false; +var should_consume = process.env.KOP_CONSUME || false; + +var counter = 0; + +if (should_consume) { + var consumer = new Kafka.KafkaConsumer({ + //'debug': 'all', + 'metadata.broker.list': broker, + 'group.id': 'node-rdkafka-consumer-flow-example', + 'enable.auto.commit': false + }, { + "auto.offset.reset": "earliest", + }); + + var topicName = topic; + + //logging debug messages, if debug is enabled + consumer.on('event.log', function (log) { + console.log(log); + }); + + //logging all errors + consumer.on('event.error', function (err) { + console.error('Error from consumer'); + console.error(err); + process.exit(1); + }); + + + consumer.on('ready', function (arg) { + console.log('consumer ready.' + JSON.stringify(arg)); + console.log("starting to consume"); + + consumer.subscribe([topicName]); + //start consuming messages + consumer.consume(); + }); + + consumer.on('data', function (m) { + + console.log("received msg", m) + + console.log('calling commit'); + consumer.commit(m); + + // Output the actual message contents + console.log(JSON.stringify(m)); + console.log(m.value.toString()); + + counter++; + + if (counter === limit) { + console.log("consumed all messages successfully"); + consumer.disconnect(); + } + }); + + consumer.on('disconnected', function (arg) { + console.log('consumer disconnected. ' + JSON.stringify(arg)); + process.exit(); + + }); + + //starting the consumer + consumer.connect(); +} + +if (should_produce) { + var producer = new Kafka.Producer({ + // 'debug' : 'all', + 'metadata.broker.list': broker, + 'dr_cb': true //delivery report callback + }); + + var topicName = topic; + + //logging debug messages, if debug is enabled + producer.on('event.log', function (log) { + console.log(log); + }); + + //logging all errors + producer.on('event.error', function (err) { + console.error('Error from producer'); + console.error(err); + process.exit(1); + }); + + producer.on('delivery-report', function (err, report) { + console.log('delivery-report: ' + JSON.stringify(report)); + counter++; + console.log(counter, limit); + if (counter === limit) { + console.log("produced all messages successfully"); + producer.disconnect(); + process.exit(); + } + }); + + //Wait for the ready event before producing + producer.on('ready', function (arg) { + console.log('producer ready.' + JSON.stringify(arg)); + console.log('starting to produce'); + + for (var i = 0; i < limit; i++) { + var value = Buffer.from('value-' + i); + var key = "key-" + i; + // if partition is set to -1, librdkafka will use the default partitioner + var partition = -1; + producer.produce(topicName, partition, value, key); + } + + //need to keep polling for a while to ensure the delivery reports are received + var pollLoop = setInterval(function () { + producer.poll(); + }, 1000); + + }); + + producer.on('disconnected', function (arg) { + console.log('producer disconnected. ' + JSON.stringify(arg)); + }); + + //starting the producer + producer.connect(); +} + diff --git a/integrations/node-rdkafka/package.json b/integrations/node-rdkafka/package.json new file mode 100644 index 0000000000..3090730a9a --- /dev/null +++ b/integrations/node-rdkafka/package.json @@ -0,0 +1,9 @@ +{ + "name": "node-rdkafka", + "version": "1.0.0", + "main": "main.js", + "license": "Apache-2.0", + "dependencies": { + "node-rdkafka": "^2.7.4" + } +} diff --git a/integrations/rustlang-rdkafka/Cargo.lock b/integrations/rustlang-rdkafka/Cargo.lock new file mode 100644 index 0000000000..66ce8efc50 --- /dev/null +++ b/integrations/rustlang-rdkafka/Cargo.lock @@ -0,0 +1,429 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +[[package]] +name = "bytes" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "cc" +version = "1.0.48" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "cfg-if" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "cmake" +version = "0.1.42" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "cc 1.0.48 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "derivative" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 0.4.30 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 0.15.44 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-channel" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-core" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-executor" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-io" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-macro" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "futures-sink" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-task" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "futures-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "memchr 2.3.0 (registry+https://github.com/rust-lang/crates.io-index)", + "pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)", + "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "itoa" +version = "0.4.4" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "libc" +version = "0.2.66" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "libz-sys" +version = "1.0.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "cc 1.0.48 (registry+https://github.com/rust-lang/crates.io-index)", + "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)", + "pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)", + "vcpkg 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "log" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "cfg-if 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "memchr" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "num_enum" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "derivative 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)", + "num_enum_derive 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "num_enum_derive" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro-crate 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)", + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "pin-project-lite" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "pin-utils" +version = "0.1.0-alpha.4" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "pkg-config" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "proc-macro-crate" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "toml 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "proc-macro-hack" +version = "0.5.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "proc-macro-nested" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "proc-macro2" +version = "0.4.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "unicode-xid 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "proc-macro2" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "quote" +version = "0.6.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 0.4.30 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "quote" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "rdkafka" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)", + "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", + "rdkafka-sys 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", + "serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", + "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "rdkafka-sys" +version = "1.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "cmake 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)", + "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)", + "libz-sys 1.0.25 (registry+https://github.com/rust-lang/crates.io-index)", + "num_enum 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", + "pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "rustlang-rdkafka" +version = "0.1.0" +dependencies = [ + "futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", + "libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)", + "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)", + "rdkafka 0.23.1 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", + "serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", + "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)", + "tokio 0.2.9 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "ryu" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "serde" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "serde_derive" +version = "1.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "serde_json" +version = "1.0.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)", + "ryu 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "slab" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "syn" +version = "0.15.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 0.4.30 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)", + "unicode-xid 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "syn" +version = "1.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)", + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "tokio" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "bytes 0.5.3 (registry+https://github.com/rust-lang/crates.io-index)", + "pin-project-lite 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)", + "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)", + "tokio-macros 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "tokio-macros" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)", + "syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "toml" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +dependencies = [ + "serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)", +] + +[[package]] +name = "unicode-xid" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "unicode-xid" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[[package]] +name = "vcpkg" +version = "0.2.8" +source = "registry+https://github.com/rust-lang/crates.io-index" + +[metadata] +"checksum bytes 0.5.3 (registry+https://github.com/rust-lang/crates.io-index)" = "10004c15deb332055f7a4a208190aed362cf9a7c2f6ab70a305fba50e1105f38" +"checksum cc 1.0.48 (registry+https://github.com/rust-lang/crates.io-index)" = "f52a465a666ca3d838ebbf08b241383421412fe7ebb463527bba275526d89f76" +"checksum cfg-if 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)" = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822" +"checksum cmake 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)" = "81fb25b677f8bf1eb325017cb6bb8452f87969db0fedb4f757b297bee78a7c62" +"checksum derivative 1.0.3 (registry+https://github.com/rust-lang/crates.io-index)" = "942ca430eef7a3806595a6737bc388bf51adb888d3fc0dd1b50f1c170167ee3a" +"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987" +"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86" +"checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866" +"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231" +"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff" +"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764" +"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16" +"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9" +"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76" +"checksum itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)" = "501266b7edd0174f8530248f87f99c88fbe60ca4ef3dd486835b8d8d53136f7f" +"checksum libc 0.2.66 (registry+https://github.com/rust-lang/crates.io-index)" = "d515b1f41455adea1313a4a2ac8a8a477634fbae63cc6100e3aebb207ce61558" +"checksum libz-sys 1.0.25 (registry+https://github.com/rust-lang/crates.io-index)" = "2eb5e43362e38e2bca2fd5f5134c4d4564a23a5c28e9b95411652021a8675ebe" +"checksum log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)" = "14b6052be84e6b71ab17edffc2eeabf5c2c3ae1fdb464aae35ac50c67a44e1f7" +"checksum memchr 2.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "3197e20c7edb283f87c071ddfc7a2cca8f8e0b888c242959846a6fce03c72223" +"checksum num_enum 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "be601e38e20a6f3d01049d85801cb9b7a34a8da7a0da70df507bbde7735058c8" +"checksum num_enum_derive 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "b59f30f6a043f2606adbd0addbf1eef6f2e28e8c4968918b63b7ff97ac0db2a7" +"checksum pin-project-lite 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "e8822eb8bb72452f038ebf6048efa02c3fe22bf83f76519c9583e47fc194a422" +"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587" +"checksum pkg-config 0.3.17 (registry+https://github.com/rust-lang/crates.io-index)" = "05da548ad6865900e60eaba7f589cc0783590a92e940c26953ff81ddbab2d677" +"checksum proc-macro-crate 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "e10d4b51f154c8a7fb96fd6dad097cb74b863943ec010ac94b9fd1be8861fe1e" +"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5" +"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e" +"checksum proc-macro2 0.4.30 (registry+https://github.com/rust-lang/crates.io-index)" = "cf3d2011ab5c909338f7887f4fc896d35932e29146c12c8d01da6b22a80ba759" +"checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27" +"checksum quote 0.6.13 (registry+https://github.com/rust-lang/crates.io-index)" = "6ce23b6b870e8f94f81fb0a363d65d86675884b34a09043c81e5562f11c1f8e1" +"checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe" +"checksum rdkafka 0.23.1 (registry+https://github.com/rust-lang/crates.io-index)" = "d455ac2a07a27d87b4f0e321dfd8d9a5b574bd96f55e42e6c594712a08051222" +"checksum rdkafka-sys 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "7d770343fbbc6089c750000711a17a906e8b3f7831afcd752d9667d38833e578" +"checksum ryu 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "bfa8506c1de11c9c4e4c38863ccbe02a305c8188e85a05a784c9e11e1c3910c8" +"checksum serde 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "414115f25f818d7dfccec8ee535d76949ae78584fc4f79a6f45a904bf8ab4449" +"checksum serde_derive 1.0.104 (registry+https://github.com/rust-lang/crates.io-index)" = "128f9e303a5a29922045a830221b8f78ec74a5f544944f3d5984f8ec3895ef64" +"checksum serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)" = "48c575e0cc52bdd09b47f330f646cf59afc586e9c4e3ccd6fc1f625b8ea1dad7" +"checksum slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "c111b5bd5695e56cffe5129854aa230b39c93a305372fdbb2668ca2394eea9f8" +"checksum syn 0.15.44 (registry+https://github.com/rust-lang/crates.io-index)" = "9ca4b3b69a77cbe1ffc9e198781b7acb0c7365a883670e8f1c1bc66fba79a5c5" +"checksum syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)" = "dff0acdb207ae2fe6d5976617f887eb1e35a2ba52c13c7234c790960cdad9238" +"checksum tokio 0.2.9 (registry+https://github.com/rust-lang/crates.io-index)" = "ffa2fdcfa937b20cb3c822a635ceecd5fc1a27a6a474527e5516aa24b8c8820a" +"checksum tokio-macros 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "50a61f268a3db2acee8dcab514efc813dc6dbe8a00e86076f935f94304b59a7a" +"checksum toml 0.5.5 (registry+https://github.com/rust-lang/crates.io-index)" = "01d1404644c8b12b16bfcffa4322403a91a451584daaaa7c28d3152e6cbc98cf" +"checksum unicode-xid 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "fc72304796d0818e357ead4e000d19c9c174ab23dc11093ac919054d20a6a7fc" +"checksum unicode-xid 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "826e7639553986605ec5979c7dd957c7895e93eabed50ab2ffa7f6128a75097c" +"checksum vcpkg 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)" = "3fc439f2794e98976c88a2a2dafce96b930fe8010b0a256b3c2199a773933168" diff --git a/integrations/rustlang-rdkafka/Cargo.toml b/integrations/rustlang-rdkafka/Cargo.toml new file mode 100644 index 0000000000..8917ec8c1f --- /dev/null +++ b/integrations/rustlang-rdkafka/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "rustlang-rdkafka" +version = "0.1.0" +authors = ["Pierre Zemb "] +edition = "2018" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +rdkafka = { version = "0.23.1", features = ["cmake-build"] } +futures = "0.3" +tokio = { version = "0.2", features = ["blocking", "macros", "rt-core", "time"] } +libc = "0.2.0" +log = "0.4.8" +serde = "1.0.0" +serde_derive = "1.0.0" +serde_json = "1.0.0" diff --git a/integrations/rustlang-rdkafka/Dockerfile b/integrations/rustlang-rdkafka/Dockerfile new file mode 100644 index 0000000000..7370767088 --- /dev/null +++ b/integrations/rustlang-rdkafka/Dockerfile @@ -0,0 +1,38 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +FROM rust:1.40-stretch + +RUN apt update +RUN apt install -y cmake libc6-dev + +WORKDIR /usr/src/rustlang-rdkafka +COPY . . + +RUN cargo install --path . + +CMD sh -c 'rustlang-rdkafka; echo "ExitCode=$?"' \ No newline at end of file diff --git a/integrations/rustlang-rdkafka/src/main.rs b/integrations/rustlang-rdkafka/src/main.rs new file mode 100644 index 0000000000..0f0620bd8f --- /dev/null +++ b/integrations/rustlang-rdkafka/src/main.rs @@ -0,0 +1,196 @@ +extern crate futures; +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// +#[macro_use] +extern crate log; +extern crate rdkafka; +extern crate tokio; + +use std::env; + +use futures::StreamExt; +use rdkafka::ClientContext; +use rdkafka::config::{ClientConfig, RDKafkaLogLevel}; +use rdkafka::consumer::{CommitMode, Consumer, ConsumerContext, Rebalance}; +use rdkafka::consumer::stream_consumer::StreamConsumer; +use rdkafka::error::KafkaResult; +use rdkafka::message::{Headers, Message}; +use rdkafka::message::OwnedHeaders; +use rdkafka::producer::{FutureProducer, FutureRecord}; +use rdkafka::topic_partition_list::TopicPartitionList; + +use crate::futures::FutureExt; + +#[tokio::main] +async fn main() -> Result<(), std::io::Error> { + let brokers = env::var("KOP_BROKER").unwrap_or_else(|_| String::from("localhost:9092")); + let topic = env::var("KOP_TOPIC").unwrap_or_else(|_| String::from("rustlang")); + let limit: i8 = env::var("KOP_LIMIT") + .unwrap_or_else(|_| String::from("10")) + .parse() + .unwrap_or(10); + let should_produce: bool = env::var("KOP_PRODUCE") + .unwrap_or_else(|_| String::from("false")) + .parse() + .unwrap_or(false); + let should_consume: bool = env::var("KOP_CONSUME") + .unwrap_or_else(|_| String::from("false")) + .parse() + .unwrap_or(false); + + println!( + "produce={}, consume={}, limit={}", + should_produce, should_consume, limit + ); + + if should_produce { + println!("starting to produce"); + produce(brokers.as_ref(), topic.as_str(), limit).await?; + } + if should_consume { + println!("starting to consume"); + consume_and_print( + brokers.as_ref(), + "rustlang-librdkafka", + vec![topic.as_str()], + limit, + ).await?; + } + println!("exiting normally"); + + Ok(()) +} + +async fn produce(brokers: &str, topic_name: &str, limit: i8) -> Result<(), std::io::Error> { + println!("starting to produce"); + let producer: FutureProducer = ClientConfig::new() + .set("bootstrap.servers", brokers) + .set("message.timeout.ms", "5000") + .create() + .expect("Producer creation error"); + + // This loop is non blocking: all messages will be sent one after the other, without waiting + // for the results. + let futures = (0..limit) + .map(|i| { + // The send operation on the topic returns a future, that will be completed once the + // result or failure from Kafka will be received. + producer + .send( + FutureRecord::to(topic_name) + .payload(&format!("Message {}", i)) + .key(&format!("Key {}", i)) + .headers(OwnedHeaders::new().add("header_key", "header_value")), + 0, + ) + .map(move |delivery_status| { + // This will be executed onw the result is received + println!("Delivery status for message {} received", i); + delivery_status + }) + }) + .collect::>(); + + // This loop will wait until all delivery statuses have been received received. + for future in futures { + println!("Future completed. Result: {:?}", future.await); + } + println!( + "produced all messages successfully ({})", + limit + ); + Ok(()) +} + +// A context can be used to change the behavior of producers and consumers by adding callbacks +// that will be executed by librdkafka. +// This particular context sets up custom callbacks to log rebalancing events. +struct CustomContext; + +impl ClientContext for CustomContext {} + +impl ConsumerContext for CustomContext { + fn pre_rebalance(&self, rebalance: &Rebalance) { + println!("Pre rebalance {:?}", rebalance); + } + + fn post_rebalance(&self, rebalance: &Rebalance) { + println!("Post rebalance {:?}", rebalance); + } + + fn commit_callback(&self, result: KafkaResult<()>, _offsets: &TopicPartitionList) { + info!("Committing offsets: {:?}", result); + } +} + +// A type alias with your custom consumer can be created for convenience. +type LoggingConsumer = StreamConsumer; + +async fn consume_and_print(brokers: &str, group_id: &str, topics: Vec<&str>, limit: i8) -> Result<(), std::io::Error> { + let context = CustomContext; + + let consumer: LoggingConsumer = ClientConfig::new() + .set("group.id", group_id) + .set("bootstrap.servers", brokers) + .set("enable.partition.eof", "false") + .set("session.timeout.ms", "6000") + .set("enable.auto.commit", "true") + //.set("statistics.interval.ms", "30000") + .set("auto.offset.reset", "earliest") + .set_log_level(RDKafkaLogLevel::Debug) + .create_with_context(context) + .expect("Consumer creation failed"); + + consumer + .subscribe(topics.as_slice()) + .expect("Can't subscribe to specified topics"); + + // consumer.start() returns a stream. The stream can be used ot chain together expensive steps, + // such as complex computations on a thread pool or asynchronous IO. + let mut message_stream = consumer.start(); + + let mut i = 0; + + while let Some(message) = message_stream.next().await { + match message { + Err(e) => panic!("Kafka error: {}", e), + Ok(m) => { + let payload = match m.payload_view::() { + None => "", + Some(Ok(s)) => s, + Some(Err(e)) => { + panic!("Error while deserializing message payload: {:?}", e); + } + }; + println!("key: '{:?}', payload: '{}', topic: {}, partition: {}, offset: {}, timestamp: {:?}", + m.key(), payload, m.topic(), m.partition(), m.offset(), m.timestamp()); + if let Some(headers) = m.headers() { + for i in 0..headers.count() { + let header = headers.get(i).unwrap(); + info!(" Header {:#?}: {:?}", header.0, header.1); + } + } + consumer.commit_message(&m, CommitMode::Sync).unwrap(); + i += 1; + println!("received msg"); + if i == limit { + println!("consumed all messages successfully"); + return Ok(()); + } + } + }; + } + Ok(()) +} diff --git a/pom.xml b/pom.xml index 600e0c2947..2afa695b98 100644 --- a/pom.xml +++ b/pom.xml @@ -212,6 +212,12 @@ test + + org.testcontainers + testcontainers + 1.12.4 + test + diff --git a/src/main/java/io/streamnative/kop/KafkaRequestHandler.java b/src/main/java/io/streamnative/kop/KafkaRequestHandler.java index 80bb47aea9..6c74ba2abd 100644 --- a/src/main/java/io/streamnative/kop/KafkaRequestHandler.java +++ b/src/main/java/io/streamnative/kop/KafkaRequestHandler.java @@ -40,6 +40,7 @@ import java.net.InetSocketAddress; import java.net.URI; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -65,6 +66,7 @@ import org.apache.commons.lang3.tuple.Pair; import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.record.MemoryRecords; import org.apache.kafka.common.record.RecordBatch; @@ -172,14 +174,88 @@ public KafkaRequestHandler(PulsarService pulsarService, kafkaConfig.getKafkaNamespace()); } + static Node newNode(InetSocketAddress address) { + if (log.isDebugEnabled()) { + log.debug("Return Broker Node of {}. {}:{}", address, address.getHostString(), address.getPort()); + } + return new Node( + Murmur3_32Hash.getInstance().makeHash((address.getHostString() + address.getPort()).getBytes(UTF_8)), + address.getHostString(), + address.getPort()); + } + + static PartitionMetadata newPartitionMetadata(TopicName topicName, Node node) { + int pulsarPartitionIndex = topicName.getPartitionIndex(); + int kafkaPartitionIndex = pulsarPartitionIndex == -1 ? 0 : pulsarPartitionIndex; + + if (log.isDebugEnabled()) { + log.debug("Return PartitionMetadata node: {}, topicName: {}", node, topicName); + } + + return new PartitionMetadata( + Errors.NONE, + kafkaPartitionIndex, + node, // leader + Lists.newArrayList(node), // replicas + Lists.newArrayList(node), // isr + Collections.emptyList() // offline replicas + ); + } + + static PartitionMetadata newFailedPartitionMetadata(TopicName topicName) { + int pulsarPartitionIndex = topicName.getPartitionIndex(); + int kafkaPartitionIndex = pulsarPartitionIndex == -1 ? 0 : pulsarPartitionIndex; + + log.warn("Failed find Broker metadata, create PartitionMetadata with INVALID_PARTITIONS"); + + return new PartitionMetadata( + Errors.UNKNOWN_SERVER_ERROR, + kafkaPartitionIndex, + Node.noNode(), // leader + Lists.newArrayList(Node.noNode()), // replicas + Lists.newArrayList(Node.noNode()), // isr + Collections.emptyList() // offline replicas + ); + } + + static AbstractResponse failedResponse(KafkaHeaderAndRequest requestHar, Throwable e) { + if (log.isDebugEnabled()) { + log.debug("Request {} get failed response ", requestHar.getHeader().apiKey(), e); + } + return requestHar.getRequest().getErrorResponse(((Integer) THROTTLE_TIME_MS.defaultValue), e); + } + protected CompletableFuture handleApiVersionsRequest(KafkaHeaderAndRequest apiVersionRequest) { - AbstractResponse apiResponse = ApiVersionsResponse.defaultApiVersionsResponse(); + AbstractResponse apiResponse = overloadDefaultApiVersionsResponse(); CompletableFuture resultFuture = new CompletableFuture<>(); resultFuture.complete(ResponseAndRequest.of(apiResponse, apiVersionRequest)); return resultFuture; } + protected ApiVersionsResponse overloadDefaultApiVersionsResponse() { + List versionList = new ArrayList<>(); + for (ApiKeys apiKey : ApiKeys.values()) { + if (apiKey.minRequiredInterBrokerMagic <= RecordBatch.CURRENT_MAGIC_VALUE) { + switch (apiKey) { + case FETCH: + // V4 added MessageSets responses. We need to make sure RecordBatch format is not used + versionList.add(new ApiVersionsResponse.ApiVersion( + (short) 1, (short) 4, apiKey.latestVersion())); + break; + case LIST_OFFSETS: + // V0 is needed for librdkafka + versionList.add(new ApiVersionsResponse.ApiVersion( + (short) 2, (short) 0, apiKey.latestVersion())); + break; + default: + versionList.add(new ApiVersionsResponse.ApiVersion(apiKey)); + } + } + } + return new ApiVersionsResponse(0, Errors.NONE, versionList); + } + protected CompletableFuture handleError(KafkaHeaderAndRequest kafkaHeaderAndRequest) { CompletableFuture resultFuture = new CompletableFuture<>(); String err = String.format("Kafka API (%s) Not supported by kop server.", @@ -375,7 +451,10 @@ protected CompletableFuture handleTopicMetadataRequest(Kafka allTopicMetadata.add( new TopicMetadata( Errors.NONE, - TopicName.get(topic).getLocalName(), + // we should answer with the right name, either local of full-name, + // depending on what was asked + topic.startsWith("persistent://") + ? TopicName.get(topic).toString() : TopicName.get(topic).getLocalName(), false, partitionMetadatas)); @@ -544,7 +623,7 @@ protected CompletableFuture handleOffsetFetchRequest(KafkaHe } private CompletableFuture - fetchOffsetForTimestamp(PersistentTopic persistentTopic, Long timestamp) { + fetchOffsetForTimestamp(PersistentTopic persistentTopic, Long timestamp, boolean legacyMode) { ManagedLedgerImpl managedLedger = (ManagedLedgerImpl) persistentTopic.getManagedLedger(); @@ -561,11 +640,20 @@ protected CompletableFuture handleOffsetFetchRequest(KafkaHe // no entry in ledger, then entry id could be -1 long entryId = position.getEntryId(); - partitionData.complete(new ListOffsetResponse.PartitionData( - Errors.NONE, - RecordBatch.NO_TIMESTAMP, - MessageIdUtils - .getOffset(position.getLedgerId(), entryId == -1 ? 0 : entryId))); + if (legacyMode) { + partitionData.complete(new ListOffsetResponse.PartitionData( + Errors.NONE, + Collections.singletonList(MessageIdUtils + .getOffset(position.getLedgerId(), entryId == -1 ? 0 : entryId)))); + + } else { + partitionData.complete(new ListOffsetResponse.PartitionData( + Errors.NONE, + RecordBatch.NO_TIMESTAMP, + MessageIdUtils + .getOffset(position.getLedgerId(), entryId == -1 ? 0 : entryId))); + } + } else if (timestamp == ListOffsetRequest.EARLIEST_TIMESTAMP) { PositionImpl position = OffsetFinder.getFirstValidPosition(managedLedger); @@ -574,10 +662,18 @@ protected CompletableFuture handleOffsetFetchRequest(KafkaHe persistentTopic.getName(), timestamp, position); } - partitionData.complete(new ListOffsetResponse.PartitionData( - Errors.NONE, - RecordBatch.NO_TIMESTAMP, - MessageIdUtils.getOffset(position.getLedgerId(), position.getEntryId()))); + if (legacyMode) { + partitionData.complete(new ListOffsetResponse.PartitionData( + Errors.NONE, + Collections.singletonList(MessageIdUtils.getOffset( + position.getLedgerId(), position.getEntryId())))); + } else { + partitionData.complete(new ListOffsetResponse.PartitionData( + Errors.NONE, + RecordBatch.NO_TIMESTAMP, + MessageIdUtils.getOffset(position.getLedgerId(), position.getEntryId()))); + } + } else { // find with real wanted timestamp OffsetFinder offsetFinder = new OffsetFinder(managedLedger); @@ -592,11 +688,18 @@ public void findEntryComplete(Position position, Object ctx) { log.warn("Unable to find position for topic {} time {}. get NULL position", persistentTopic.getName(), timestamp); - partitionData.complete(new ListOffsetResponse - .PartitionData( - Errors.UNKNOWN_SERVER_ERROR, - ListOffsetResponse.UNKNOWN_TIMESTAMP, - ListOffsetResponse.UNKNOWN_OFFSET)); + if (legacyMode) { + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + Collections.emptyList())); + } else { + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + ListOffsetResponse.UNKNOWN_TIMESTAMP, + ListOffsetResponse.UNKNOWN_OFFSET)); + } return; } } else { @@ -607,10 +710,17 @@ public void findEntryComplete(Position position, Object ctx) { log.debug("Find position for topic {} time {}. position: {}", persistentTopic.getName(), timestamp, finalPosition); } - partitionData.complete(new ListOffsetResponse.PartitionData( - Errors.NONE, - RecordBatch.NO_TIMESTAMP, - MessageIdUtils.getOffset(finalPosition.getLedgerId(), finalPosition.getEntryId()))); + if (legacyMode) { + partitionData.complete(new ListOffsetResponse.PartitionData( + Errors.NONE, + Collections.singletonList(MessageIdUtils.getOffset( + finalPosition.getLedgerId(), finalPosition.getEntryId())))); + } else { + partitionData.complete(new ListOffsetResponse.PartitionData( + Errors.NONE, + RecordBatch.NO_TIMESTAMP, + MessageIdUtils.getOffset(finalPosition.getLedgerId(), finalPosition.getEntryId()))); + } } @Override @@ -618,11 +728,18 @@ public void findEntryFailed(ManagedLedgerException exception, Optional position, Object ctx) { log.warn("Unable to find position for topic {} time {}. Exception:", persistentTopic.getName(), timestamp, exception); - partitionData.complete(new ListOffsetResponse - .PartitionData( - Errors.UNKNOWN_SERVER_ERROR, - ListOffsetResponse.UNKNOWN_TIMESTAMP, - ListOffsetResponse.UNKNOWN_OFFSET)); + if (legacyMode) { + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + Collections.emptyList())); + } else { + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + ListOffsetResponse.UNKNOWN_TIMESTAMP, + ListOffsetResponse.UNKNOWN_OFFSET)); + } return; } }); @@ -630,16 +747,79 @@ public void findEntryFailed(ManagedLedgerException exception, } catch (Exception e) { log.error("Failed while get position for topic: {} ts: {}.", persistentTopic.getName(), timestamp, e); - - partitionData.complete(new ListOffsetResponse.PartitionData( - Errors.UNKNOWN_SERVER_ERROR, - ListOffsetResponse.UNKNOWN_TIMESTAMP, - ListOffsetResponse.UNKNOWN_OFFSET)); + if (legacyMode) { + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + Collections.emptyList())); + } else { + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + ListOffsetResponse.UNKNOWN_TIMESTAMP, + ListOffsetResponse.UNKNOWN_OFFSET)); + } } - return partitionData; } + // Some info can be found here: + // https://cfchou.github.io/blog/2015/04/23/a-closer-look-at-kafka-offsetrequest/ + // Site is available through https://web.archive.org/ + private CompletableFuture handleListOffsetRequestV0(KafkaHeaderAndRequest listOffset) { + ListOffsetRequest request = (ListOffsetRequest) listOffset.getRequest(); + + CompletableFuture resultFuture = new CompletableFuture<>(); + Map> responseData = Maps.newHashMap(); + + // in v0, the iterator is offsetData, + // in v1, the iterator is partitionTimestamps, + log.warn("received a v0 listOffset: {}", request.toString(true)); + request.offsetData().entrySet().stream().forEach(tms -> { + TopicPartition topic = tms.getKey(); + TopicName pulsarTopic = pulsarTopicName(topic, namespace); + Long times = tms.getValue().timestamp; + CompletableFuture partitionData; + + // num_num_offsets > 1 is not handled for now, returning an error + if (tms.getValue().maxNumOffsets > 1) { + log.warn("request is asking for multiples offsets for {}, not supported for now", + pulsarTopic.toString()); + partitionData = new CompletableFuture<>(); + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_SERVER_ERROR, + Collections.singletonList(ListOffsetResponse.UNKNOWN_OFFSET))); + } + + // topic not exist, return UNKNOWN_TOPIC_OR_PARTITION + if (!topicManager.topicExists(pulsarTopic.toString())) { + log.warn("Topic {} not exist in topic manager while list offset.", pulsarTopic.toString()); + partitionData = new CompletableFuture<>(); + partitionData.complete(new ListOffsetResponse + .PartitionData( + Errors.UNKNOWN_TOPIC_OR_PARTITION, + Collections.singletonList(ListOffsetResponse.UNKNOWN_OFFSET))); + } else { + PersistentTopic persistentTopic = topicManager.getTopic(pulsarTopic.toString()); + partitionData = fetchOffsetForTimestamp(persistentTopic, times, true); + } + responseData.put(topic, partitionData); + }); + + CompletableFuture + .allOf(responseData.values().stream().toArray(CompletableFuture[]::new)) + .whenComplete((ignore, ex) -> { + ListOffsetResponse response = + new ListOffsetResponse(CoreUtils.mapValue(responseData, future -> future.join())); + + resultFuture.complete(ResponseAndRequest + .of(response, listOffset)); + }); + + return resultFuture; + } + private CompletableFuture handleListOffsetRequestV1AndAbove(KafkaHeaderAndRequest listOffset) { ListOffsetRequest request = (ListOffsetRequest) listOffset.getRequest(); @@ -663,7 +843,7 @@ private CompletableFuture handleListOffsetRequestV1AndAbove( ListOffsetResponse.UNKNOWN_OFFSET)); } else { PersistentTopic persistentTopic = topicManager.getTopic(pulsarTopic.toString()); - partitionData = fetchOffsetForTimestamp(persistentTopic, times); + partitionData = fetchOffsetForTimestamp(persistentTopic, times, false); } responseData.put(topic, partitionData); @@ -685,24 +865,11 @@ private CompletableFuture handleListOffsetRequestV1AndAbove( // get offset from underline managedLedger protected CompletableFuture handleListOffsetRequest(KafkaHeaderAndRequest listOffset) { checkArgument(listOffset.getRequest() instanceof ListOffsetRequest); - - // not support version 0 + // the only difference between v0 and v1 is the `max_num_offsets => INT32` + // v0 is required because it is used by librdkafka if (listOffset.getHeader().apiVersion() == 0) { - CompletableFuture resultFuture = new CompletableFuture<>(); - ListOffsetRequest request = (ListOffsetRequest) listOffset.getRequest(); - - log.error("ListOffset not support V0 format request"); - - ListOffsetResponse response = new ListOffsetResponse(CoreUtils.mapValue(request.partitionTimestamps(), - ignored -> new ListOffsetResponse - .PartitionData(Errors.UNSUPPORTED_FOR_MESSAGE_FORMAT, Lists.newArrayList()))); - - resultFuture.complete(ResponseAndRequest - .of(response, listOffset)); - - return resultFuture; + return handleListOffsetRequestV0(listOffset); } - return handleListOffsetRequestV1AndAbove(listOffset); } @@ -1172,16 +1339,6 @@ private CompletableFuture findBroker(PulsarService pulsarServ } } - static Node newNode(InetSocketAddress address) { - if (log.isDebugEnabled()) { - log.debug("Return Broker Node of {}. {}:{}", address, address.getHostString(), address.getPort()); - } - return new Node( - Murmur3_32Hash.getInstance().makeHash((address.getHostString() + address.getPort()).getBytes(UTF_8)), - address.getHostString(), - address.getPort()); - } - Node newSelfNode() { String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress( kafkaConfig.getAdvertisedAddress()); @@ -1197,46 +1354,4 @@ Node newSelfNode() { hostname, port); } - - - static PartitionMetadata newPartitionMetadata(TopicName topicName, Node node) { - int pulsarPartitionIndex = topicName.getPartitionIndex(); - int kafkaPartitionIndex = pulsarPartitionIndex == -1 ? 0 : pulsarPartitionIndex; - - if (log.isDebugEnabled()) { - log.debug("Return PartitionMetadata node: {}, topicName: {}", node, topicName); - } - - return new PartitionMetadata( - Errors.NONE, - kafkaPartitionIndex, - node, // leader - Lists.newArrayList(node), // replicas - Lists.newArrayList(node), // isr - Collections.emptyList() // offline replicas - ); - } - - static PartitionMetadata newFailedPartitionMetadata(TopicName topicName) { - int pulsarPartitionIndex = topicName.getPartitionIndex(); - int kafkaPartitionIndex = pulsarPartitionIndex == -1 ? 0 : pulsarPartitionIndex; - - log.warn("Failed find Broker metadata, create PartitionMetadata with INVALID_PARTITIONS"); - - return new PartitionMetadata( - Errors.UNKNOWN_SERVER_ERROR, - kafkaPartitionIndex, - Node.noNode(), // leader - Lists.newArrayList(Node.noNode()), // replicas - Lists.newArrayList(Node.noNode()), // isr - Collections.emptyList() // offline replicas - ); - } - - static AbstractResponse failedResponse(KafkaHeaderAndRequest requestHar, Throwable e) { - if (log.isDebugEnabled()) { - log.debug("Request {} get failed response ", requestHar.getHeader().apiKey(), e); - } - return requestHar.getRequest().getErrorResponse(((Integer) THROTTLE_TIME_MS.defaultValue), e); - } } diff --git a/src/test/java/io/streamnative/kop/KafkaIntegrationTest.java b/src/test/java/io/streamnative/kop/KafkaIntegrationTest.java new file mode 100644 index 0000000000..a4bf956161 --- /dev/null +++ b/src/test/java/io/streamnative/kop/KafkaIntegrationTest.java @@ -0,0 +1,207 @@ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.streamnative.kop; + +import static org.testng.AssertJUnit.assertFalse; +import static org.testng.AssertJUnit.assertTrue; + +import com.google.common.collect.Sets; +import java.nio.file.Paths; +import java.util.Optional; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.RetentionPolicies; +import org.apache.pulsar.common.policies.data.TenantInfo; +import org.junit.AfterClass; +import org.testcontainers.Testcontainers; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.output.WaitingConsumer; +import org.testcontainers.containers.wait.strategy.Wait; +import org.testcontainers.images.builder.ImageFromDockerfile; +import org.testcontainers.shaded.com.google.common.collect.ImmutableMap; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +/** + * This class is running Integration tests for Kafka clients. + * It uses testcontainers to spawn containers that will either: + * * produce a number of messages + * * consume a number of messages + * + *

As testcontainers is not capable of checking exitCode of the app running in the container, + * Every container should print the exitCode in stdout. + * + *

This class is waiting for some precise logs to come-up: + * * "ready to produce" + * * "ready to consume" + * * "produced all messages successfully" + * * "consumed all messages successfully" + * + *

This class is using environment variables to control the containers, such as: + * * broker address, + * * topic name, + * * produce or consume mode, + * * how many message to produce/consume, + */ +@Slf4j +public class KafkaIntegrationTest extends MockKafkaServiceBaseTest { + + @DataProvider + public static Object[][] integrations() { + return new Object[][]{ + {"golang-sarama", Optional.empty(), true, true}, + {"golang-sarama", Optional.of("persistent://public/default/my-sarama-topic-full-name"), true, true}, + {"golang-confluent-kafka", Optional.empty(), true, true}, + {"rustlang-rdkafka", Optional.empty(), true, true}, + // consumer is broken, see integrations/README.md + {"node-kafka-node", Optional.empty(), true, false}, + {"node-rdkafka", Optional.empty(), true, true}, + }; + } + + private static WaitingConsumer createLogFollower(final GenericContainer container) { + final WaitingConsumer waitingConsumer = new WaitingConsumer(); + container.followOutput(waitingConsumer); + return waitingConsumer; + } + + private static void checkForErrorsInLogs(final String logs) { + assertFalse(logs.contains("no available broker to send metadata request to")); + assertFalse(logs.contains("panic")); + assertFalse(logs.contains("correlation ID didn't match")); + assertFalse(logs.contains("Required feature not supported by broker")); + + + if (logs.contains("starting to produce")) { + assertTrue(logs.contains("produced all messages successfully")); + } + + if (logs.contains("starting to consume")) { + assertTrue(logs.contains("consumed all messages successfully")); + } + assertTrue(logs.contains("ExitCode=0")); + } + + @BeforeClass + @Override + protected void setup() throws Exception { + + super.resetConfig(); + super.internalSetup(); + + if (!this.admin.clusters().getClusters().contains(this.configClusterName)) { + // so that clients can test short names + this.admin.clusters().createCluster(this.configClusterName, + new ClusterData("http://127.0.0.1:" + this.brokerWebservicePort)); + } else { + this.admin.clusters().updateCluster(this.configClusterName, + new ClusterData("http://127.0.0.1:" + this.brokerWebservicePort)); + } + + if (!this.admin.tenants().getTenants().contains("public")) { + this.admin.tenants().createTenant("public", + new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test"))); + } else { + this.admin.tenants().updateTenant("public", + new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test"))); + } + if (!this.admin.namespaces().getNamespaces("public").contains("public/default")) { + this.admin.namespaces().createNamespace("public/default"); + this.admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test")); + this.admin.namespaces().setRetention("public/default", + new RetentionPolicies(60, 1000)); + } + if (!this.admin.namespaces().getNamespaces("public").contains("public/__kafka")) { + this.admin.namespaces().createNamespace("public/__kafka"); + this.admin.namespaces().setNamespaceReplicationClusters("public/__kafka", Sets.newHashSet("test")); + this.admin.namespaces().setRetention("public/__kafka", + new RetentionPolicies(-1, -1)); + } + Testcontainers.exposeHostPorts(ImmutableMap.of(super.kafkaBrokerPort, super.kafkaBrokerPort)); + } + + @Test(timeOut = 3 * 60_000, dataProvider = "integrations") + void simpleProduceAndConsume(final String integration, final Optional topic, + final boolean shouldProduce, final boolean shouldConsume) throws Exception { + + this.getAdmin().topics().createPartitionedTopic(topic.orElse(integration), 1); + + final GenericContainer producer = new GenericContainer<>( + new ImageFromDockerfile().withFileFromPath(".", Paths.get("integrations/" + integration))) + .withEnv("KOP_BROKER", "localhost:" + super.kafkaBrokerPort) + .withEnv("KOP_PRODUCE", "true") + .withEnv("KOP_TOPIC", topic.orElse(integration)) + .withEnv("KOP_LIMIT", "10") + .withLogConsumer(new org.testcontainers.containers.output.Slf4jLogConsumer(KafkaIntegrationTest.log)) + .waitingFor(Wait.forLogMessage("starting to produce\\n", 1)) + .withNetworkMode("host"); + + final GenericContainer consumer = new GenericContainer<>( + new ImageFromDockerfile().withFileFromPath(".", Paths.get("integrations/" + integration))) + .withEnv("KOP_BROKER", "localhost:" + super.kafkaBrokerPort) + .withEnv("KOP_TOPIC", topic.orElse(integration)) + .withEnv("KOP_CONSUME", "true") + .withEnv("KOP_LIMIT", "10") + .withLogConsumer(new org.testcontainers.containers.output.Slf4jLogConsumer(KafkaIntegrationTest.log)) + .waitingFor(Wait.forLogMessage("starting to consume\\n", 1)) + .withNetworkMode("host"); + + WaitingConsumer producerWaitingConsumer = null; + WaitingConsumer consumerWaitingConsumer = null; + if (shouldProduce) { + producer.start(); + producerWaitingConsumer = KafkaIntegrationTest.createLogFollower(producer); + System.out.println("producer started"); + } + + if (shouldConsume) { + consumer.start(); + consumerWaitingConsumer = KafkaIntegrationTest.createLogFollower(consumer); + System.out.println("consumer started"); + } + + if (shouldProduce) { + producerWaitingConsumer.waitUntil(frame -> + frame.getUtf8String().contains("ExitCode"), 30, TimeUnit.SECONDS); + KafkaIntegrationTest.checkForErrorsInLogs(producer.getLogs()); + } + + if (shouldConsume) { + consumerWaitingConsumer.waitUntil(frame -> + frame.getUtf8String().contains("ExitCode"), 30, TimeUnit.SECONDS); + KafkaIntegrationTest.checkForErrorsInLogs(consumer.getLogs()); + } + } + + @Override + @AfterClass + protected void cleanup() throws Exception { + super.internalCleanup(); + } +} diff --git a/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java b/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java index 652577db88..e07982f827 100644 --- a/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java +++ b/src/test/java/io/streamnative/kop/MockKafkaServiceBaseTest.java @@ -105,7 +105,6 @@ protected void resetConfig() { this.conf.setAdvertisedAddress("localhost"); this.conf.setWebServicePort(Optional.ofNullable(brokerWebservicePort)); this.conf.setClusterName(configClusterName); - this.conf.setAdvertisedAddress("localhost"); this.conf.setListeners( PLAINTEXT_PREFIX + "localhost:" + kafkaBrokerPort + "," + SSL_PREFIX + "localhost:" + kafkaBrokerPortTls);