[GitHub] [beam] methodmissing commented on pull request #12794: [BEAM-10865] Support for Kafka deserialization API with headers (since Kafka API 2.1.0)

[GitBox]

methodmissing commented on pull request #12794:
URL: https://github.com/apache/beam/pull/12794#issuecomment-695482531


   Hi Luke,
   
   I got it working with a resolution strategy to force `kafka-clients:2.1.0` 
for the `kafkaVersion210` configuration. It's not super clean and I'd prefer to 
not have to use it, but I think this dependency `provided 
library.java.kafka_clients` is hard to substitute otherwise.
   
   Original Kafka test, default API:
   
   <img width="1486" alt="Screenshot 2020-09-20 at 01 45 31" 
src="https://user-images.githubusercontent.com/379/93692551-8fcca200-faec-11ea-992e-1b61cb12396f.png";>
   
   API 2.1.0 specific test:
   
   <img width="1201" alt="Screenshot 2020-09-20 at 01 46 00" 
src="https://user-images.githubusercontent.com/379/93692562-9bb86400-faec-11ea-96d8-6e27bff4da58.png";>
   
   I reverse tested with the following corruption of the 2.1.0 path in the 
`ConsumerSpEL` to confirm API 2.1.0 was loaded at runtime:
   
   ```diff
   diff --git 
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
 
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
   index 9ca5bfc990..37a4a2bc8e 100644
   --- 
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
   +++ 
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
   @@ -132,7 +132,7 @@ class ConsumerSpEL {
        mapContext.setVariable("deserializer", deserializer);
        mapContext.setVariable("topic", rawRecord.topic());
        mapContext.setVariable("headers", rawRecord.headers());
   -    mapContext.setVariable("data", isKey ? rawRecord.key() : 
rawRecord.value());
   +    mapContext.setVariable("data", isKey ? rawRecord.value() : 
rawRecord.key());
        return deserializeWithHeadersExpression.getValue(mapContext);
      }
   ```


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org