KIP-42 Add producer and consumer interceptors #1
Conversation
* Functional tests are skipped except when the `functional` build tag is passed to the go test command (i.e. go test -tags=functional) * If TOXIPROXY_ADDR is not set when the functional tests are being run, it will use docker-compose to automatically spin up a zookeeper/kafka/toxiproxy environment suitab le for running the tests. This requies a working Docker and for the docker-compose command line tool to be installed. * If TOXIPROXY_ADDR and KAFKA_VERSION are set, then the tests will not spin up any docker infrastructure and will instead rely on whatever kafka broker is behind the specified toxiproxy.
I think we should add the onClose since we use it to close the trace exporter |
still unsure about this. |
Include zstd on the functional tests
| // ProducerInterceptor allows you to intercept (and possibly mutate) the records | ||
| // received by the producer before they are published to the Kafka cluster. | ||
| // https://cwiki.apache.org/confluence/display/KAFKA/KIP-42%3A+Add+Producer+and+Consumer+Interceptors#KIP42:AddProducerandConsumerInterceptors-Motivation | ||
| type ProducerInterceptor interface { |
There was a problem hiding this comment.
do we need a ProducerInterceptor and ConsumerInterceptor? could we just have an Interceptor with an onCall(message)?
There was a problem hiding this comment.
we would need at least 2 methods, one that receives a ProducerMessage and other for ConsumerMessage, and that might be an issue, as if someone would like to only intercept a producer, it'll need to implement the onSend for consumer too.
So I think it's safe to leave it in 2 separated interfaces
functional_producer_test.go
Outdated
| @@ -25,6 +26,12 @@ func TestFuncProducingGzip(t *testing.T) { | |||
| testProducingMessages(t, config) | |||
| } | |||
|
|
|||
| func TestFuncProducingZstd(t *testing.T) { | |||
There was a problem hiding this comment.
maybe rebase since IBM#1726 was merged
interceptors.go
Outdated
| // onConsume is called when the consumed message is intercepted. Please | ||
| // avoid modifying the message until it's safe to do so, as this is _not_ a | ||
| // copy of the message. |
There was a problem hiding this comment.
this is a bit scary since monitoring should not affect the code. this doc implies that an error in monitoring could break the code
There was a problem hiding this comment.
yeah, that was my opinionated take on this.
I added some comments on the upstream PR about this, let's see if we can get some 👀 on this.
There was a problem hiding this comment.
this doc implies that an error in monitoring could break the code
Not break the code, but modify partially the message
d1egoaz
force-pushed
the
diego_interceptors
branch
4 times, most recently
from
5664afe
to
4c63884
Compare
…_tests Use docker-compose to run the functional tests
- expose a `Commit()` sync method on ConsumerGroupSession - don't create mainLoop in OffsetManager unless AutoCommit is enabled
Consumer group support for manually comitting offsets
This PR includes: Producer: `onSend` but it doesn't implement `onAcknowledgement` Consumer: `onConsume` but it doesn't implement `onCommit` I'm not sure if I need to add the `onClose` method. Maybe in another iteration ¯\_(ツ)_/¯
d1egoaz
force-pushed
the
diego_interceptors
branch
from
f10663f
to
a299948
Compare
This PR includes:
Producer:
onSendbut it doesn't implementonAcknowledgementConsumer:
onConsumebut it doesn't implementonCommitI'm not sure if I need to add the
onClosemethod. Maybe in anotheriteration ¯_(ツ)_/¯
TODO: functional test, but it looks the way sarama functional test works is that they don't have this detail.