Support gRPC client stream/unary oneway #1384
Conversation
...fka.spec/src/test/java/io/aklivity/zilla/specs/binding/grpc/kafka/streams/GrpcProduceIT.java
Outdated
Show resolved
Hide resolved
...fka.spec/src/test/java/io/aklivity/zilla/specs/binding/grpc/kafka/streams/GrpcProduceIT.java
Outdated
Show resolved
Hide resolved
...ka.spec/src/test/java/io/aklivity/zilla/specs/binding/grpc/kafka/streams/KafkaProduceIT.java
Outdated
Show resolved
Hide resolved
...ka.spec/src/test/java/io/aklivity/zilla/specs/binding/grpc/kafka/streams/KafkaProduceIT.java
Outdated
Show resolved
Hide resolved
...ity/zilla/specs/binding/grpc/kafka/streams/kafka/produce/client.stream.rpc.oneway/client.rpt
Outdated
Show resolved
Hide resolved
...io/aklivity/zilla/runtime/binding/grpc/kafka/internal/config/GrpcKafkaWithProduceResult.java
Outdated
Show resolved
Hide resolved
...java/io/aklivity/zilla/runtime/binding/grpc/kafka/internal/stream/GrpcKafkaProxyFactory.java
Outdated
Show resolved
Hide resolved
...java/io/aklivity/zilla/runtime/binding/grpc/kafka/internal/stream/GrpcKafkaProxyFactory.java
Show resolved
Hide resolved
...ity/zilla/specs/binding/grpc/kafka/streams/kafka/produce/client.stream.rpc.oneway/server.rpt
Outdated
Show resolved
Hide resolved
| protected void onKafkaData( | ||
| long traceId, | ||
| long authorization, | ||
| long budgetId, | ||
| int reserved, | ||
| int flags, | ||
| OctetsFW payload, | ||
| KafkaDataExFW kafkaDataEx) | ||
| { | ||
| } |
There was a problem hiding this comment.
Please follow the same design as HttpKafkaProduceNoReply, where the onKafkaData method in super type is non-abstract and empty implementation. Then subclasses are not forced to override the method where it is not needed.
Note: AFAIK, KafkaProduceProxy does not actually call onKafkaData on HttpProxy delegate.
| @@ -1707,7 +2116,8 @@ private void doKafkaData( | |||
|
|
|||
| private void doKafkaEnd( | |||
| long traceId, | |||
| long authorization) | |||
| long authorization, | |||
| boolean tombstone) | |||
There was a problem hiding this comment.
Let's avoid changing the signature of doKafkaEnd to handle two different use cases that are not needed by the same implementation instance.
Instead, our overall design approach here has been to use strategy objects dedicated to each use case, adjusting the behavior as needed in each implementation class.
Suggest introducing a KafkaProduceNoReplyProxy that is a subclass of KafkaProduceProxy where doKafkaEnd is protected and overridden in the subclass to get the desired behavior.
Then the type of the delegate in the Grpc stream remains KafkaProduceProxy but instantiates KafkaProduceProxy for GrpcProduceProxy and KafkaProduceNoReplyProxy for GrpcProduceNoReplyProxy. In each case calling delegate.doKafkaEnd(traceId, authorization) causes the desired behavior due to the method override.
Fixes #642