diff --git a/CHANGELOG.md b/CHANGELOG.md index 01a0a28..fc61f16 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,26 +6,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] -## 5.0.6 - 2024-03-14 - -## 5.0.6 - 2024-03-14 +## 5.1.0 - 2024-08-07 -## 5.0.6 - 2024-03-14 +### Added +- [#207](https://github.com/deviceinsight/kafkactl/pull/207) Allow configuring TLS for avro schema registry +- [#193](https://github.com/deviceinsight/kafkactl/pull/193) Print group instance IDs in `describe consumer-group` command ## 5.0.6 - 2024-03-14 -## 5.0.5 - 2024-03-12 - -## 5.0.4 - 2024-03-12 - -## 5.0.3 - 2024-03-12 - -## 5.0.2 - 2024-03-08 - -## 5.0.1 - 2024-03-08 - -## 5.0.0 - 2024-03-08 - ### Added - [#190](https://github.com/deviceinsight/kafkactl/pull/190) Improve handling of project config files - [#192](https://github.com/deviceinsight/kafkactl/pull/192) Plugin infrastructure for tokenProviders diff --git a/README.adoc b/README.adoc index 48a1265..db9a00b 100644 --- a/README.adoc +++ b/README.adoc @@ -24,13 +24,6 @@ You can install the pre-compiled binary or compile from source. === Install the pre-compiled binary -*snap*: - -[,bash] ----- -snap install kafkactl ----- - *homebrew*: [,bash] @@ -43,6 +36,12 @@ brew install deviceinsight/packages/kafkactl brew upgrade deviceinsight/packages/kafkactl ---- +*winget*: +[,bash] +---- +winget install kafkactl +---- + *deb/rpm*: Download the .deb or .rpm from the https://github.com/deviceinsight/kafkactl/releases[releases page] and install with dpkg -i and rpm -i respectively. @@ -51,6 +50,13 @@ Download the .deb or .rpm from the https://github.com/deviceinsight/kafkactl/rel There's a kafkactl https://aur.archlinux.org/packages/kafkactl/[AUR package] available for Arch. Install it with your AUR helper of choice (e.g. https://github.com/Jguer/yay[yay]): +*snap*: + +[,bash] +---- +snap install kafkactl +---- + [,bash] ---- yay -S kafkactl @@ -64,7 +70,7 @@ Download the pre-compiled binaries from the https://github.com/deviceinsight/kaf [,bash] ---- -go get -u github.com/deviceinsight/kafkactl +go get -u github.com/deviceinsight/kafkactl/v5 ---- *NOTE:* make sure that `kafkactl` is on PATH otherwise auto-completion won't work. @@ -104,8 +110,15 @@ contexts: enabled: true username: admin password: admin - # optional configure sasl mechanism as plaintext, scram-sha256, scram-sha512 (defaults to plaintext) - mechanism: scram-sha512 + # optional configure sasl mechanism as plaintext, scram-sha256, scram-sha512, oauth (defaults to plaintext) + mechanism: oauth + # optional tokenProvider configuration (only used for 'sasl.mechanism=oauth') + tokenprovider: + # plugin to use as token provider implementation (see plugin section) + plugin: azure + # optional: additional options passed to the plugin + options: + key: value # optional: access clusters running kubernetes kubernetes: @@ -147,6 +160,18 @@ contexts: # see: https://github.com/deviceinsight/kafkactl/issues/123 jsonCodec: avro + # optional: timeout for requests (defaults to 5s) + requestTimeout: 10s + + # optional: tls config for avro + tls: + enabled: true + ca: my-ca + cert: my-cert + certKey: my-key + # set insecure to true to ignore all tls verification (defaults to false) + insecure: false + # optional: default protobuf messages search paths protobuf: importPaths: @@ -185,6 +210,7 @@ The config file location is resolved by . as default the config file is looked up from one of the following locations: ** `$HOME/.config/kafkactl/config.yml` ** `$HOME/.kafkactl/config.yml` + ** `$APPDATA/kafkactl/config.yml` ** `$SNAP_REAL_HOME/.kafkactl/config.yml` ** `$SNAP_DATA/kafkactl/config.yml` ** `/etc/kafkactl/config.yml` @@ -359,6 +385,19 @@ If environment variables for the `default` context should be set, the prefix `CO So, instead of `CONTEXTS_DEFAULT_TLS_CERTKEY` one can also set `TLS_CERTKEY`. See *root_test.go* for more examples. +== Plugins + +_kafkactl_ supports plugins to cope with specifics when using Kafka-compatible clusters available from cloud providers such as Azure or AWS. + +At the moment, plugins can only be used to implement a `tokenProvider` for _oauth_ authentication. +In the future, plugins might implement additional commands to query data or configuration which is not part of the Kafka-API. One example would be Eventhub consumer groups/offsets for Azure. + +See the plugin documentation for additional documentation and usage examples. + +Available plugins: + +* https://github.com/deviceinsight/kafkactl-plugins/blob/main/azure/README.adoc[azure plugin] + == Examples === Consuming messages @@ -795,7 +834,7 @@ The assigned replicas of a partition can directly be altered with: [,bash] ---- # set brokers 102,103 as replicas for partition 3 of topic my-topic -kafkactl alter topic my-topic 3 -r 102,103 +kafkactl alter partition my-topic 3 -r 102,103 ---- === Clone topic diff --git a/docker/run-integration-tests.sh b/docker/run-integration-tests.sh index 905213d..d88a96e 100755 --- a/docker/run-integration-tests.sh +++ b/docker/run-integration-tests.sh @@ -1,7 +1,7 @@ #!/bin/bash # use: # export NO_DOCKER_COMPOSE=true -# to skip docker-compose when it is already running locally +# to skip docker compose when it is already running locally set -e set -o pipefail @@ -14,7 +14,7 @@ echo "using kafka version: ${KAFKAVERSION}" # docker compose up pushd ${ROOT_DIR} if [[ -z "${NO_DOCKER_COMPOSE}" ]]; then - docker-compose -f ${SCRIPT_DIR}/docker-compose.yml --env-file=${SCRIPT_DIR}/.env up -d + docker compose -f ${SCRIPT_DIR}/docker-compose.yml --env-file=${SCRIPT_DIR}/.env up -d fi @@ -22,7 +22,7 @@ fi function tearDown { popd >/dev/null 2>&1 if [[ -z "${NO_DOCKER_COMPOSE}" ]]; then - docker-compose -f ${SCRIPT_DIR}/docker-compose.yml --env-file=${SCRIPT_DIR}/.env down + docker compose -f ${SCRIPT_DIR}/docker-compose.yml --env-file=${SCRIPT_DIR}/.env down fi } trap tearDown EXIT diff --git a/go.mod b/go.mod index 8820fd7..9808d92 100644 --- a/go.mod +++ b/go.mod @@ -1,38 +1,38 @@ module github.com/deviceinsight/kafkactl/v5 -go 1.21.6 +go 1.22.6 require ( - github.com/IBM/sarama v1.43.0 + github.com/IBM/sarama v1.43.2 github.com/Rican7/retry v0.3.1 github.com/golang/protobuf v1.5.4 - github.com/hashicorp/go-hclog v1.6.2 - github.com/hashicorp/go-plugin v1.6.0 - github.com/jhump/protoreflect v1.15.6 - github.com/landoop/schema-registry v0.0.0-20190327143759-50a5701c1891 - github.com/linkedin/goavro/v2 v2.12.0 + github.com/hashicorp/go-hclog v1.6.3 + github.com/hashicorp/go-plugin v1.6.1 + github.com/jhump/protoreflect v1.16.0 + github.com/linkedin/goavro/v2 v2.13.0 github.com/pkg/errors v0.9.1 - github.com/spf13/cobra v1.8.0 + github.com/riferrei/srclient v0.6.0 + github.com/spf13/cobra v1.8.1 github.com/spf13/pflag v1.0.5 - github.com/spf13/viper v1.18.2 + github.com/spf13/viper v1.19.0 github.com/xdg-go/scram v1.1.2 go.uber.org/ratelimit v0.3.1 - golang.org/x/sync v0.6.0 - golang.org/x/term v0.18.0 - google.golang.org/protobuf v1.33.0 + golang.org/x/sync v0.8.0 + golang.org/x/term v0.23.0 + google.golang.org/protobuf v1.34.2 gopkg.in/errgo.v2 v2.1.0 gopkg.in/yaml.v2 v2.4.0 ) require ( github.com/benbjohnson/clock v1.3.5 // indirect - github.com/bufbuild/protocompile v0.9.0 // indirect - github.com/cpuguy83/go-md2man/v2 v2.0.3 // indirect + github.com/bufbuild/protocompile v0.14.0 // indirect + github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect - github.com/eapache/go-resiliency v1.6.0 // indirect + github.com/eapache/go-resiliency v1.7.0 // indirect github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect github.com/eapache/queue v1.1.0 // indirect - github.com/fatih/color v1.16.0 // indirect + github.com/fatih/color v1.17.0 // indirect github.com/fsnotify/fsnotify v1.7.0 // indirect github.com/golang/snappy v0.0.4 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect @@ -46,35 +46,36 @@ require ( github.com/jcmturner/gofork v1.7.6 // indirect github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect github.com/jcmturner/rpc/v2 v2.0.3 // indirect - github.com/klauspost/compress v1.17.7 // indirect + github.com/klauspost/compress v1.17.9 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/mitchellh/go-testing-interface v1.14.1 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/oklog/run v1.1.0 // indirect - github.com/pelletier/go-toml/v2 v2.1.1 // indirect + github.com/pelletier/go-toml/v2 v2.2.2 // indirect github.com/pierrec/lz4/v4 v4.1.21 // indirect github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect github.com/rogpeppe/go-internal v1.12.0 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect - github.com/sagikazarmark/locafero v0.4.0 // indirect + github.com/sagikazarmark/locafero v0.6.0 // indirect github.com/sagikazarmark/slog-shim v0.1.0 // indirect + github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 // indirect github.com/sourcegraph/conc v0.3.0 // indirect github.com/spf13/afero v1.11.0 // indirect - github.com/spf13/cast v1.6.0 // indirect + github.com/spf13/cast v1.7.0 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/xdg-go/pbkdf2 v1.0.0 // indirect github.com/xdg-go/stringprep v1.0.4 // indirect go.uber.org/atomic v1.11.0 // indirect go.uber.org/multierr v1.11.0 // indirect - golang.org/x/crypto v0.21.0 // indirect - golang.org/x/exp v0.0.0-20240222234643-814bf88cf225 // indirect - golang.org/x/net v0.22.0 // indirect - golang.org/x/sys v0.18.0 // indirect - golang.org/x/text v0.14.0 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20240304212257-790db918fca8 // indirect - google.golang.org/grpc v1.62.1 // indirect + golang.org/x/crypto v0.26.0 // indirect + golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 // indirect + golang.org/x/net v0.28.0 // indirect + golang.org/x/sys v0.23.0 // indirect + golang.org/x/text v0.17.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20240805194559-2c9e96a0b5d4 // indirect + google.golang.org/grpc v1.65.0 // indirect gopkg.in/ini.v1 v1.67.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/go.sum b/go.sum index 859f203..ed11140 100644 --- a/go.sum +++ b/go.sum @@ -1,26 +1,26 @@ -github.com/IBM/sarama v1.43.0 h1:YFFDn8mMI2QL0wOrG0J2sFoVIAFl7hS9JQi2YZsXtJc= -github.com/IBM/sarama v1.43.0/go.mod h1:zlE6HEbC/SMQ9mhEYaF7nNLYOUyrs0obySKCckWP9BM= +github.com/IBM/sarama v1.43.2 h1:HABeEqRUh32z8yzY2hGB/j8mHSzC/HA9zlEjqFNCzSw= +github.com/IBM/sarama v1.43.2/go.mod h1:Kyo4WkF24Z+1nz7xeVUFWIuKVV8RS3wM8mkvPKMdXFQ= github.com/Rican7/retry v0.3.1 h1:scY4IbO8swckzoA/11HgBwaZRJEyY9vaNJshcdhp1Mc= github.com/Rican7/retry v0.3.1/go.mod h1:CxSDrhAyXmTMeEuRAnArMu1FHu48vtfjLREWqVl7Vw0= github.com/benbjohnson/clock v1.3.5 h1:VvXlSJBzZpA/zum6Sj74hxwYI2DIxRWuNIoXAzHZz5o= github.com/benbjohnson/clock v1.3.5/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= -github.com/bufbuild/protocompile v0.9.0 h1:DI8qLG5PEO0Mu1Oj51YFPqtx6I3qYXUAhJVJ/IzAVl0= -github.com/bufbuild/protocompile v0.9.0/go.mod h1:s89m1O8CqSYpyE/YaSGtg1r1YFMF5nLTwh4vlj6O444= -github.com/cpuguy83/go-md2man/v2 v2.0.3 h1:qMCsGGgs+MAzDFyp9LpAe1Lqy/fY/qCovCm0qnXZOBM= -github.com/cpuguy83/go-md2man/v2 v2.0.3/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o= +github.com/bufbuild/protocompile v0.14.0 h1:z3DW4IvXE5G/uTOnSQn+qwQQxvhckkTWLS/0No/o7KU= +github.com/bufbuild/protocompile v0.14.0/go.mod h1:N6J1NYzkspJo3ZwyL4Xjvli86XOj1xq4qAasUFxGups= +github.com/cpuguy83/go-md2man/v2 v2.0.4 h1:wfIWP927BUkWJb2NmU/kNDYIBTh/ziUX91+lVfRxZq4= +github.com/cpuguy83/go-md2man/v2 v2.0.4/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/eapache/go-resiliency v1.6.0 h1:CqGDTLtpwuWKn6Nj3uNUdflaq+/kIPsg0gfNzHton30= -github.com/eapache/go-resiliency v1.6.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho= +github.com/eapache/go-resiliency v1.7.0 h1:n3NRTnBn5N0Cbi/IeOHuQn9s2UwVUH7Ga0ZWcP+9JTA= +github.com/eapache/go-resiliency v1.7.0/go.mod h1:5yPzW0MIvSe0JDsv0v+DvcjEv2FyD6iZYSs1ZI+iQho= github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 h1:Oy0F4ALJ04o5Qqpdz8XLIpNA3WM/iSIXqxtqo7UGVws= github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3/go.mod h1:YvSRo5mw33fLEx1+DlK6L2VV43tJt5Eyel9n9XBcR+0= github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc= github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I= github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk= -github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM= -github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE= +github.com/fatih/color v1.17.0 h1:GlRw1BRJxkpqUCBKzKOw098ed57fEsKeNjpTe3cSjK4= +github.com/fatih/color v1.17.0/go.mod h1:YZ7TlrGPkiz6ku9fK3TLD/pl3CpsiFyu8N92HLgmosI= github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw= github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g= github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= @@ -39,12 +39,12 @@ github.com/gorilla/sessions v1.2.1/go.mod h1:dk2InVEVJ0sfLlnXv9EAgkf6ecYs/i80K/z github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I= github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4= -github.com/hashicorp/go-hclog v1.6.2 h1:NOtoftovWkDheyUM/8JW3QMiXyxJK3uHRK7wV04nD2I= -github.com/hashicorp/go-hclog v1.6.2/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= +github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB11/k= +github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo= github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM= -github.com/hashicorp/go-plugin v1.6.0 h1:wgd4KxHJTVGGqWBq4QPB1i5BZNEx9BR8+OFmHDmTk8A= -github.com/hashicorp/go-plugin v1.6.0/go.mod h1:lBS5MtSSBZk0SHc66KACcjjlU6WzEVP/8pwz68aMkCI= +github.com/hashicorp/go-plugin v1.6.1 h1:P7MR2UP6gNKGPp+y7EZw2kOiq4IR9WiqLvp0XOsVdwI= +github.com/hashicorp/go-plugin v1.6.1/go.mod h1:XPHFku2tFo3o3QKFgSYo+cghcUhw1NA1hZyMK0PWAw0= github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8= github.com/hashicorp/go-uuid v1.0.3/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro= @@ -66,10 +66,10 @@ github.com/jcmturner/gokrb5/v8 v8.4.4 h1:x1Sv4HaTpepFkXbt2IkL29DXRf8sOfZXo8eRKh6 github.com/jcmturner/gokrb5/v8 v8.4.4/go.mod h1:1btQEpgT6k+unzCwX1KdWMEwPPkkgBtP+F6aCACiMrs= github.com/jcmturner/rpc/v2 v2.0.3 h1:7FXXj8Ti1IaVFpSAziCZWNzbNuZmnvw/i6CqLNdWfZY= github.com/jcmturner/rpc/v2 v2.0.3/go.mod h1:VUJYCIDm3PVOEHw8sgt091/20OJjskO/YJki3ELg/Hc= -github.com/jhump/protoreflect v1.15.6 h1:WMYJbw2Wo+KOWwZFvgY0jMoVHM6i4XIvRs2RcBj5VmI= -github.com/jhump/protoreflect v1.15.6/go.mod h1:jCHoyYQIJnaabEYnbGwyo9hUqfyUMTbJw/tAut5t97E= -github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg= -github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/jhump/protoreflect v1.16.0 h1:54fZg+49widqXYQ0b+usAFHbMkBGR4PpXrsHc8+TBDg= +github.com/jhump/protoreflect v1.16.0/go.mod h1:oYPd7nPvcBw/5wlDfm/AVmU9zH9BgqGCI469pGxfj/8= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -77,10 +77,9 @@ 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/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/landoop/schema-registry v0.0.0-20190327143759-50a5701c1891 h1:FADDInPE0OtV85SKuJAGwcTiXwzyg2ztBqtUWA5EF04= -github.com/landoop/schema-registry v0.0.0-20190327143759-50a5701c1891/go.mod h1:yITyTTMx2IS5mpfZjQ64gJhL5U5RvcorFBu+z4/euXg= -github.com/linkedin/goavro/v2 v2.12.0 h1:rIQQSj8jdAUlKQh6DttK8wCRv4t4QO09g1C4aBWXslg= -github.com/linkedin/goavro/v2 v2.12.0/go.mod h1:KXx+erlq+RPlGSPmLF7xGo6SAbh8sCQ53x064+ioxhk= +github.com/linkedin/goavro/v2 v2.11.1/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA= +github.com/linkedin/goavro/v2 v2.13.0 h1:L8eI8GcuciwUkt41Ej62joSZS4kKaYIUdze+6for9NU= +github.com/linkedin/goavro/v2 v2.13.0/go.mod h1:KXx+erlq+RPlGSPmLF7xGo6SAbh8sCQ53x064+ioxhk= github.com/magiconair/properties v1.8.7 h1:IeQXZAiQcpL9mgcAe1Nu6cX9LLw6ExEHKjN0VQdvPDY= github.com/magiconair/properties v1.8.7/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc= @@ -98,8 +97,8 @@ github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyua github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/oklog/run v1.1.0 h1:GEenZ1cK0+q0+wsJew9qUg/DyD8k3JzYsZAi5gYi2mA= github.com/oklog/run v1.1.0/go.mod h1:sVPdnTZT1zYwAJeCMu2Th4T21pA3FPOQRfWjQlk7DVU= -github.com/pelletier/go-toml/v2 v2.1.1 h1:LWAJwfNvjQZCFIDKWYQaM62NcYeYViCmWIwmOStowAI= -github.com/pelletier/go-toml/v2 v2.1.1/go.mod h1:tJU2Z3ZkXwnxa4DPO899bsyIoywizdUvyaeZurnPPDc= +github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM= +github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs= github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ= github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= @@ -109,37 +108,44 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRI github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM= github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4= +github.com/riferrei/srclient v0.6.0 h1:60LWpQW66AAL5TtWuMPZEplwgWLUdCK3OBUbag/JWFg= +github.com/riferrei/srclient v0.6.0/go.mod h1:e3nZcDdaOSsaYqiO18INPBK4qnJTjEEyL2rlJcsTtrA= github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8= github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4= github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= -github.com/sagikazarmark/locafero v0.4.0 h1:HApY1R9zGo4DBgr7dqsTH/JJxLTTsOt7u6keLGt6kNQ= -github.com/sagikazarmark/locafero v0.4.0/go.mod h1:Pe1W6UlPYUk/+wc/6KFhbORCfqzgYEpgQ3O5fPuL3H4= +github.com/sagikazarmark/locafero v0.6.0 h1:ON7AQg37yzcRPU69mt7gwhFEBwxI6P9T4Qu3N51bwOk= +github.com/sagikazarmark/locafero v0.6.0/go.mod h1:77OmuIc6VTraTXKXIs/uvUxKGUXjE1GbemJYHqdNjX0= github.com/sagikazarmark/slog-shim v0.1.0 h1:diDBnUNK9N/354PgrxMywXnAwEr1QZcOr6gto+ugjYE= github.com/sagikazarmark/slog-shim v0.1.0/go.mod h1:SrcSrq8aKtyuqEI1uvTDTK1arOWRIczQRv+GVI1AkeQ= +github.com/santhosh-tekuri/jsonschema/v5 v5.0.0/go.mod h1:FKdcjfQW6rpZSnxxUvEA5H/cDPdvJ/SZJQLWWXWGrZ0= +github.com/santhosh-tekuri/jsonschema/v5 v5.3.1 h1:lZUw3E0/J3roVtGQ+SCrUrg3ON6NgVqpn3+iol9aGu4= +github.com/santhosh-tekuri/jsonschema/v5 v5.3.1/go.mod h1:uToXkOrWAZ6/Oc07xWQrPOhJotwFIyu2bBVN41fcDUY= github.com/sourcegraph/conc v0.3.0 h1:OQTbbt6P72L20UqAkXXuLOj79LfEanQ+YQFNpLA9ySo= github.com/sourcegraph/conc v0.3.0/go.mod h1:Sdozi7LEKbFPqYX2/J+iBAM6HpqSLTASQIKqDmF7Mt0= github.com/spf13/afero v1.11.0 h1:WJQKhtpdm3v2IzqG8VMqrr6Rf3UYpEF239Jy9wNepM8= github.com/spf13/afero v1.11.0/go.mod h1:GH9Y3pIexgf1MTIWtNGyogA5MwRIDXGUr+hbWNoBjkY= -github.com/spf13/cast v1.6.0 h1:GEiTHELF+vaR5dhz3VqZfFSzZjYbgeKDpBxQVS4GYJ0= -github.com/spf13/cast v1.6.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= -github.com/spf13/cobra v1.8.0 h1:7aJaZx1B85qltLMc546zn58BxxfZdR/W22ej9CFoEf0= -github.com/spf13/cobra v1.8.0/go.mod h1:WXLWApfZ71AjXPya3WOlMsY9yMs7YeiHhFVlvLyhcho= +github.com/spf13/cast v1.7.0 h1:ntdiHjuueXFgm5nzDRdOS4yfT43P5Fnud6DH50rz/7w= +github.com/spf13/cast v1.7.0/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo= +github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM= +github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y= github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/spf13/viper v1.18.2 h1:LUXCnvUvSM6FXAsj6nnfc8Q2tp1dIgUfY9Kc8GsSOiQ= -github.com/spf13/viper v1.18.2/go.mod h1:EKmWIqdnk5lOcmR72yw6hS+8OPYcwD0jteitLMVB+yk= +github.com/spf13/viper v1.19.0 h1:RWq5SEjt8o25SROyN3z2OrDB9l7RPd3lwTWU8EcEdcI= +github.com/spf13/viper v1.19.0/go.mod h1:GQUN9bilAbhU/jgc1bKs99f/suXKeUMct8Adx5+Ntkg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1FQKckRals= github.com/stretchr/testify v1.7.5/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= -github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= @@ -158,10 +164,10 @@ go.uber.org/ratelimit v0.3.1/go.mod h1:6euWsTB6U/Nb3X++xEUXA8ciPJvr19Q/0h1+oDcJh golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= -golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA= -golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs= -golang.org/x/exp v0.0.0-20240222234643-814bf88cf225 h1:LfspQV/FYTatPTr/3HzIcmiUFH7PGP+OQ6mgDYo3yuQ= -golang.org/x/exp v0.0.0-20240222234643-814bf88cf225/go.mod h1:CxmFvTBINI24O/j8iY7H1xHzx2i4OsyguNBmN/uPtqc= +golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw= +golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54= +golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 h1:2dVuKD2vS7b0QIHQbpyTISPd0LeHDbnYEryqj5Q1ug8= +golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56/go.mod h1:M4RDyNAINzryxdtnbRXRL/OHtkFuWGRjvuhBJpk2IlY= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= @@ -169,12 +175,13 @@ golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.22.0 h1:9sGLhx7iRIHEiX0oAJ3MRZMUCElJgy7Br1nO+AMN3Tc= -golang.org/x/net v0.22.0/go.mod h1:JKghWKKOSdJwpW2GEx0Ja7fmaKnMsbu+MWVZTokSYmg= +golang.org/x/net v0.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE= +golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220513210516-0976fa681c29/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ= -golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ= +golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -188,30 +195,30 @@ golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4= -golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM= +golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k= -golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8= -golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58= +golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU= +golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= -golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= -golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/text v0.17.0 h1:XtiM5bkSOt+ewxlOE/aE/AKEHibwj/6gvWMl9Rsh0Qc= +golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -google.golang.org/genproto/googleapis/rpc v0.0.0-20240304212257-790db918fca8 h1:IR+hp6ypxjH24bkMfEJ0yHR21+gwPWdV+/IBrPQyn3k= -google.golang.org/genproto/googleapis/rpc v0.0.0-20240304212257-790db918fca8/go.mod h1:UCOku4NytXMJuLQE5VuqA5lX3PcHCBo8pxNyvkf4xBs= -google.golang.org/grpc v1.62.1 h1:B4n+nfKzOICUXMgyrNd19h/I9oH0L1pizfk1d4zSgTk= -google.golang.org/grpc v1.62.1/go.mod h1:IWTG0VlJLCh1SkC58F7np9ka9mx/WNkjl4PGJaiq+QE= -google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= -google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240805194559-2c9e96a0b5d4 h1:OsSGQeIIsyOEOimVxLEIL4rwGcnrjOydQaiA2bOnZUM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20240805194559-2c9e96a0b5d4/go.mod h1:Ue6ibwXGpU+dqIcODieyLOcgj7z8+IcskoNIgZxtrFY= +google.golang.org/grpc v1.65.0 h1:bs/cUb4lp1G5iImFFd3u5ixQzweKizoZJAwBNLR42lc= +google.golang.org/grpc v1.65.0/go.mod h1:WgYC2ypjlB0EiQi6wdKixMqukr6lBc0Vo+oOgjrM5ZQ= +google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg= +google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= diff --git a/internal/common-operation.go b/internal/common-operation.go index 6a3fb05..9e25755 100644 --- a/internal/common-operation.go +++ b/internal/common-operation.go @@ -4,12 +4,15 @@ import ( "crypto/tls" "crypto/x509" "fmt" + "net/http" "os" "os/user" "regexp" "strings" "time" + "github.com/riferrei/srclient" + "github.com/deviceinsight/kafkactl/v5/internal/auth" "github.com/deviceinsight/kafkactl/v5/internal/global" @@ -85,6 +88,8 @@ type ClientContext struct { KafkaVersion sarama.KafkaVersion AvroSchemaRegistry string AvroJSONCodec avro.JSONCodec + AvroRequestTimeout time.Duration + AvroTLS TLSConfig Protobuf protobuf.SearchContext Producer ProducerConfig Consumer ConsumerConfig @@ -128,6 +133,12 @@ func CreateClientContext() (ClientContext, error) { } context.AvroSchemaRegistry = viper.GetString("contexts." + context.Name + ".avro.schemaRegistry") context.AvroJSONCodec = avro.ParseJSONCodec(viper.GetString("contexts." + context.Name + ".avro.jsonCodec")) + context.AvroRequestTimeout = viper.GetDuration("contexts." + context.Name + ".avro.requestTimeout") + context.AvroTLS.Enabled = viper.GetBool("contexts." + context.Name + ".avro.tls.enabled") + context.AvroTLS.CA = viper.GetString("contexts." + context.Name + ".avro.tls.ca") + context.AvroTLS.Cert = viper.GetString("contexts." + context.Name + ".avro.tls.cert") + context.AvroTLS.CertKey = viper.GetString("contexts." + context.Name + ".avro.tls.certKey") + context.AvroTLS.Insecure = viper.GetBool("contexts." + context.Name + ".avro.tls.insecure") context.Protobuf.ProtosetFiles = viper.GetStringSlice("contexts." + context.Name + ".protobuf.protosetFiles") context.Protobuf.ProtoImportPaths = viper.GetStringSlice("contexts." + context.Name + ".protobuf.importPaths") context.Protobuf.ProtoFiles = viper.GetStringSlice("contexts." + context.Name + ".protobuf.protoFiles") @@ -241,6 +252,33 @@ func CreateClientConfig(context *ClientContext) (*sarama.Config, error) { return config, nil } +func CreateAvroSchemaRegistryClient(context *ClientContext) (srclient.ISchemaRegistryClient, error) { + + timeout := context.AvroRequestTimeout + + if context.AvroRequestTimeout <= 0 { + timeout = 5 * time.Second + } + + client := &http.Client{Timeout: timeout} + + if context.AvroTLS.Enabled { + output.Debugf("avro TLS is enabled.") + + tlsConfig, err := setupTLSConfig(context.AvroTLS) + if err != nil { + return nil, errors.Wrap(err, "failed to setup avro tls config") + } + + client.Transport = &http.Transport{ + TLSClientConfig: tlsConfig, + } + } + + baseURL := avro.FormatBaseURL(context.AvroSchemaRegistry) + return srclient.CreateSchemaRegistryClientWithOptions(baseURL, client, 16), nil +} + func GetClientID(context *ClientContext, defaultPrefix string) string { var ( diff --git a/internal/consume/AvroMessageDeserializer.go b/internal/consume/AvroMessageDeserializer.go index 041e2fe..6bfe516 100644 --- a/internal/consume/AvroMessageDeserializer.go +++ b/internal/consume/AvroMessageDeserializer.go @@ -16,25 +16,9 @@ import ( ) type AvroMessageDeserializer struct { - topic string - avroSchemaRegistry string - jsonCodec avro.JSONCodec - registry *CachingSchemaRegistry -} - -func CreateAvroMessageDeserializer(topic string, avroSchemaRegistry string, jsonCodec avro.JSONCodec) (AvroMessageDeserializer, error) { - - var err error - - deserializer := AvroMessageDeserializer{topic: topic, avroSchemaRegistry: avroSchemaRegistry, jsonCodec: jsonCodec} - - deserializer.registry, err = CreateCachingSchemaRegistry(deserializer.avroSchemaRegistry) - - if err != nil { - return deserializer, errors.Wrap(err, "failed to create schema registry client: ") - } - - return deserializer, nil + topic string + jsonCodec avro.JSONCodec + registry *CachingSchemaRegistry } type avroMessage struct { diff --git a/internal/consume/CachingSchemaRegistry.go b/internal/consume/CachingSchemaRegistry.go index 0dd2368..29996cc 100644 --- a/internal/consume/CachingSchemaRegistry.go +++ b/internal/consume/CachingSchemaRegistry.go @@ -1,37 +1,24 @@ package consume import ( - schemaregistry "github.com/landoop/schema-registry" - "github.com/pkg/errors" + "github.com/riferrei/srclient" ) type CachingSchemaRegistry struct { subjects []string schemas map[int]string - client *schemaregistry.Client + client srclient.ISchemaRegistryClient } -func CreateCachingSchemaRegistry(avroSchemaRegistry string) (*CachingSchemaRegistry, error) { - - var err error - - registry := &CachingSchemaRegistry{} - - registry.schemas = make(map[int]string) - registry.client, err = schemaregistry.NewClient(avroSchemaRegistry) - - if err != nil { - return registry, errors.Wrap(err, "failed to create schema registry registry: ") - } - - return registry, nil +func CreateCachingSchemaRegistry(client srclient.ISchemaRegistryClient) *CachingSchemaRegistry { + return &CachingSchemaRegistry{client: client, schemas: make(map[int]string)} } func (registry *CachingSchemaRegistry) Subjects() ([]string, error) { var err error if len(registry.subjects) == 0 { - registry.subjects, err = registry.client.Subjects() + registry.subjects, err = registry.client.GetSubjects() } return registry.subjects, err @@ -41,10 +28,10 @@ func (registry *CachingSchemaRegistry) GetSchemaByID(id int) (string, error) { var err error if _, ok := registry.schemas[id]; !ok { - var schema string - schema, err = registry.client.GetSchemaByID(id) - if err == nil && schema != "" { - registry.schemas[id] = schema + var schema *srclient.Schema + schema, err = registry.client.GetSchema(id) + if err == nil { + registry.schemas[id] = schema.Schema() } } diff --git a/internal/consume/consume-operation.go b/internal/consume/consume-operation.go index 5cccd98..5867a5c 100644 --- a/internal/consume/consume-operation.go +++ b/internal/consume/consume-operation.go @@ -91,11 +91,14 @@ func (operation *Operation) Consume(topic string, flags Flags) error { var deserializers MessageDeserializerChain if clientContext.AvroSchemaRegistry != "" { - deserializer, err := CreateAvroMessageDeserializer(topic, clientContext.AvroSchemaRegistry, clientContext.AvroJSONCodec) + client, err := internal.CreateAvroSchemaRegistryClient(&clientContext) if err != nil { return err } + deserializer := AvroMessageDeserializer{topic: topic, registry: CreateCachingSchemaRegistry(client), + jsonCodec: clientContext.AvroJSONCodec} + deserializers = append(deserializers, deserializer) } diff --git a/internal/consumergroups/consumer-group-operation.go b/internal/consumergroups/consumer-group-operation.go index 61f99fb..e54b8cd 100644 --- a/internal/consumergroups/consumer-group-operation.go +++ b/internal/consumergroups/consumer-group-operation.go @@ -43,6 +43,7 @@ type partitionOffset struct { type consumerGroupMember struct { ClientHost string `json:"clientHost" yaml:"clientHost"` ClientID string `json:"clientId" yaml:"clientId"` + GroupInstanceID string `json:"groupInstanceId,omitempty" yaml:"groupInstanceId,omitempty"` AssignedPartitions []topicPartition `json:"assignedPartitions" yaml:"assignedPartitions"` } @@ -73,11 +74,12 @@ type ConsumerGroupOperation struct { func (operation *ConsumerGroupOperation) DescribeConsumerGroup(flags DescribeConsumerGroupFlags, group string) error { var ( - err error - ctx internal.ClientContext - client sarama.Client - admin sarama.ClusterAdmin - descriptions []*sarama.GroupDescription + err error + ctx internal.ClientContext + client sarama.Client + admin sarama.ClusterAdmin + descriptions []*sarama.GroupDescription + supportsGroupInstanceID bool ) if ctx, err = internal.CreateClientContext(); err != nil { @@ -88,6 +90,8 @@ func (operation *ConsumerGroupOperation) DescribeConsumerGroup(flags DescribeCon return errors.Wrap(err, "failed to create client") } + supportsGroupInstanceID = client.Config().Version.IsAtLeast(sarama.V2_4_0_0) + if admin, err = internal.CreateClusterAdmin(&ctx); err != nil { return errors.Wrap(err, "failed to create cluster admin") } @@ -138,7 +142,7 @@ func (operation *ConsumerGroupOperation) DescribeConsumerGroup(flags DescribeCon assignedPartitions := filterAssignedPartitions(memberAssignment.Topics, topicPartitions) - consumerGroupDescription.Members = addMember(consumerGroupDescription.Members, member.ClientHost, member.ClientId, assignedPartitions) + consumerGroupDescription.Members = addMember(consumerGroupDescription.Members, member.ClientHost, member.ClientId, member.GroupInstanceId, assignedPartitions) } sort.Slice(consumerGroupDescription.Members, func(i, j int) bool { @@ -173,14 +177,30 @@ func (operation *ConsumerGroupOperation) DescribeConsumerGroup(flags DescribeCon consumerGroupDescription.Members = nil } else if flags.OutputFormat == "wide" || flags.OutputFormat == "" { tableWriter := output.CreateTableWriter() - if err := tableWriter.WriteHeader("CLIENT_HOST", "CLIENT_ID", "TOPIC", "ASSIGNED_PARTITIONS"); err != nil { + + columns := make([]string, 0, 5) + columns = append(columns, "CLIENT_HOST", "CLIENT_ID") + if supportsGroupInstanceID { + columns = append(columns, "GROUP_INSTANCE_ID") + } + columns = append(columns, "TOPIC", "ASSIGNED_PARTITIONS") + + if err := tableWriter.WriteHeader(columns...); err != nil { return err } for _, m := range consumerGroupDescription.Members { for _, topic := range m.AssignedPartitions { partitions := strings.Trim(strings.Join(strings.Fields(fmt.Sprint(topic.Partitions)), ","), "[]") - if err := tableWriter.Write(m.ClientHost, m.ClientID, topic.Name, partitions); err != nil { + + columns = columns[:0] + columns = append(columns, m.ClientHost, m.ClientID) + if supportsGroupInstanceID { + columns = append(columns, m.GroupInstanceID) + } + columns = append(columns, topic.Name, partitions) + + if err := tableWriter.Write(columns...); err != nil { return err } } @@ -226,7 +246,7 @@ func filterAssignedPartitions(assignedPartitions map[string][]int32, topicPartit return result } -func addMember(members []consumerGroupMember, clientHost string, clientID string, assignedPartitions map[string][]int32) []consumerGroupMember { +func addMember(members []consumerGroupMember, clientHost string, clientID string, groupInstanceID *string, assignedPartitions map[string][]int32) []consumerGroupMember { topicPartitionList := make([]topicPartition, 0) @@ -247,7 +267,13 @@ func addMember(members []consumerGroupMember, clientHost string, clientID string if len(assignedPartitions) == 0 { return members } - member := consumerGroupMember{ClientHost: clientHost, ClientID: clientID, AssignedPartitions: topicPartitionList} + + groupInstanceIDString := "" + if groupInstanceID != nil { + groupInstanceIDString = *groupInstanceID + } + + member := consumerGroupMember{ClientHost: clientHost, ClientID: clientID, GroupInstanceID: groupInstanceIDString, AssignedPartitions: topicPartitionList} return append(members, member) } diff --git a/internal/helpers/avro/SchemaRegistry.go b/internal/helpers/avro/SchemaRegistry.go new file mode 100644 index 0000000..cc01c18 --- /dev/null +++ b/internal/helpers/avro/SchemaRegistry.go @@ -0,0 +1,49 @@ +package avro + +import ( + "strings" +) + +// FormatBaseURL will try to make sure that the schema:host:port pattern is followed on the `baseURL` field. +func FormatBaseURL(baseURL string) string { + if baseURL == "" { + return "" + } + + // remove last slash, so the API can append the path with ease. + if baseURL[len(baseURL)-1] == '/' { + baseURL = baseURL[0 : len(baseURL)-1] + } + + portIdx := strings.LastIndexByte(baseURL, ':') + + schemaIdx := strings.Index(baseURL, "://") + hasSchema := schemaIdx >= 0 + hasPort := portIdx > schemaIdx+1 + + var port = "80" + if hasPort { + port = baseURL[portIdx+1:] + } + + // find the schema based on the port. + if !hasSchema { + if port == "443" { + baseURL = "https://" + baseURL + } else { + baseURL = "http://" + baseURL + } + } else if !hasPort { + // has schema but not port. + if strings.HasPrefix(baseURL, "https://") { + port = "443" + } + } + + // finally, append the port part if it wasn't there. + if !hasPort { + baseURL += ":" + port + } + + return baseURL +} diff --git a/internal/producer/AvroMessageSerializer.go b/internal/producer/AvroMessageSerializer.go index eb3cbac..7b1180e 100644 --- a/internal/producer/AvroMessageSerializer.go +++ b/internal/producer/AvroMessageSerializer.go @@ -4,41 +4,25 @@ import ( "encoding/binary" "github.com/deviceinsight/kafkactl/v5/internal/helpers/avro" + "github.com/riferrei/srclient" "github.com/IBM/sarama" "github.com/deviceinsight/kafkactl/v5/internal/util" - schemaregistry "github.com/landoop/schema-registry" "github.com/linkedin/goavro/v2" "github.com/pkg/errors" ) type AvroMessageSerializer struct { - topic string - avroSchemaRegistry string - jsonCodec avro.JSONCodec - client *schemaregistry.Client -} - -func CreateAvroMessageSerializer(topic string, avroSchemaRegistry string, jsonCodec avro.JSONCodec) (AvroMessageSerializer, error) { - - var err error - - serializer := AvroMessageSerializer{topic: topic, avroSchemaRegistry: avroSchemaRegistry, jsonCodec: jsonCodec} - - serializer.client, err = schemaregistry.NewClient(serializer.avroSchemaRegistry) - - if err != nil { - return serializer, errors.Wrap(err, "failed to create schema registry client: ") - } - - return serializer, nil + topic string + jsonCodec avro.JSONCodec + client srclient.ISchemaRegistryClient } func (serializer AvroMessageSerializer) encode(rawData []byte, schemaVersion int, avroSchemaType string) ([]byte, error) { subject := serializer.topic + "-" + avroSchemaType - subjects, err := serializer.client.Subjects() + subjects, err := serializer.client.GetSubjects() if err != nil { return nil, errors.Wrap(err, "failed to list available avro schemas") @@ -49,7 +33,7 @@ func (serializer AvroMessageSerializer) encode(rawData []byte, schemaVersion int return rawData, nil } - var schema schemaregistry.Schema + var schema *srclient.Schema if schemaVersion == -1 { schema, err = serializer.client.GetLatestSchema(subject) @@ -58,7 +42,7 @@ func (serializer AvroMessageSerializer) encode(rawData []byte, schemaVersion int return nil, errors.Errorf("failed to find latest avro schema for subject: %s (%v)", subject, err) } } else { - schema, err = serializer.client.GetSchemaBySubject(subject, schemaVersion) + schema, err = serializer.client.GetSchemaByVersion(subject, schemaVersion) if err != nil { return nil, errors.Errorf("failed to find avro schema for subject: %s id: %d (%v)", subject, schemaVersion, err) @@ -68,9 +52,9 @@ func (serializer AvroMessageSerializer) encode(rawData []byte, schemaVersion int var avroCodec *goavro.Codec if serializer.jsonCodec == avro.Avro { - avroCodec, err = goavro.NewCodec(schema.Schema) + avroCodec, err = goavro.NewCodec(schema.Schema()) } else { - avroCodec, err = goavro.NewCodecForStandardJSONFull(schema.Schema) + avroCodec, err = goavro.NewCodecForStandardJSONFull(schema.Schema()) } if err != nil { @@ -89,14 +73,14 @@ func (serializer AvroMessageSerializer) encode(rawData []byte, schemaVersion int // https://docs.confluent.io/current/schema-registry/docs/serializer-formatter.html#wire-format versionBytes := make([]byte, 5) - binary.BigEndian.PutUint32(versionBytes[1:], uint32(schema.ID)) + binary.BigEndian.PutUint32(versionBytes[1:], uint32(schema.ID())) return append(versionBytes, data...), nil } func (serializer AvroMessageSerializer) CanSerialize(topic string) (bool, error) { - subjects, err := serializer.client.Subjects() + subjects, err := serializer.client.GetSubjects() if err != nil { return false, errors.Wrap(err, "failed to list available avro schemas") diff --git a/internal/producer/producer-operation.go b/internal/producer/producer-operation.go index c702307..4d3ea86 100644 --- a/internal/producer/producer-operation.go +++ b/internal/producer/producer-operation.go @@ -80,10 +80,11 @@ func (operation *Operation) Produce(topic string, flags Flags) error { serializers := MessageSerializerChain{topic: topic} if clientContext.AvroSchemaRegistry != "" { - serializer, err := CreateAvroMessageSerializer(topic, clientContext.AvroSchemaRegistry, clientContext.AvroJSONCodec) + client, err := internal.CreateAvroSchemaRegistryClient(&clientContext) if err != nil { return err } + serializer := AvroMessageSerializer{topic: topic, client: client, jsonCodec: clientContext.AvroJSONCodec} serializers.serializers = append(serializers.serializers, serializer) } diff --git a/internal/testutil/helpers.go b/internal/testutil/helpers.go index 1e3f4a3..7a99c89 100644 --- a/internal/testutil/helpers.go +++ b/internal/testutil/helpers.go @@ -9,11 +9,12 @@ import ( "testing" "time" + "github.com/riferrei/srclient" + "github.com/Rican7/retry" "github.com/Rican7/retry/backoff" "github.com/Rican7/retry/strategy" "github.com/deviceinsight/kafkactl/v5/internal/util" - schemaregistry "github.com/landoop/schema-registry" ) func CreateTopic(t *testing.T, topicPrefix string, flags ...string) string { @@ -38,20 +39,16 @@ func CreateAvroTopic(t *testing.T, topicPrefix, keySchema, valueSchema string, f topicName := CreateTopic(t, topicPrefix, flags...) - schemaRegistry, err := schemaregistry.NewClient("localhost:18081") - - if err != nil { - t.Fatalf("failed to create schema registry client: %v", err) - } + schemaRegistry := srclient.CreateSchemaRegistryClient("http://localhost:18081") if keySchema != "" { - if _, err := schemaRegistry.RegisterNewSchema(topicName+"-key", keySchema); err != nil { + if _, err := schemaRegistry.CreateSchema(topicName+"-key", keySchema, srclient.Avro); err != nil { t.Fatalf("unable to register schema for key: %v", err) } } if valueSchema != "" { - if _, err := schemaRegistry.RegisterNewSchema(topicName+"-value", valueSchema); err != nil { + if _, err := schemaRegistry.CreateSchema(topicName+"-value", valueSchema, srclient.Avro); err != nil { t.Fatalf("unable to register schema for value: %v", err) } } diff --git a/internal/util/plugin.go b/internal/util/plugin.go index 7ed5da7..9552262 100644 --- a/internal/util/plugin.go +++ b/internal/util/plugin.go @@ -70,6 +70,19 @@ func resolvePluginPath(pluginName string) (string, error) { pluginExecutable := fmt.Sprintf("kafkactl-%s-plugin%s", pluginName, extension) + // search relative to working dir + workingDir, err := os.Getwd() + if err != nil { + return "", err + } + + pluginPath := filepath.Join(workingDir, "../kafkactl-plugins/"+pluginName) + pluginLocationWorkingDir := filepath.Join(pluginPath, pluginExecutable) + + if _, err = os.Stat(pluginLocationWorkingDir); err == nil { + return pluginLocationWorkingDir, nil + } + // search in path if _, err := exec.LookPath(pluginExecutable); err == nil { return pluginExecutable, nil @@ -88,18 +101,5 @@ func resolvePluginPath(pluginName string) (string, error) { return pluginLocationExe, nil } - // search relative to working dir - workingDir, err := os.Getwd() - if err != nil { - return "", err - } - - pluginPath := filepath.Join(workingDir, "../kafkactl-plugins/"+pluginName) - pluginLocationWorkingDir := filepath.Join(pluginPath, pluginExecutable) - - if _, err = os.Stat(pluginLocationWorkingDir); err == nil { - return pluginLocationWorkingDir, nil - } - return "", errors.Wrapf(err, "plugin not found: %q", []string{pluginExecutable, pluginLocationExe, pluginLocationWorkingDir}) }