[ https://issues.apache.org/jira/browse/BEAM-3925?focusedWorklogId=93366&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-93366 ]
ASF GitHub Bot logged work on BEAM-3925: ---------------------------------------- Author: ASF GitHub Bot Created on: 20/Apr/18 18:51 Start Date: 20/Apr/18 18:51 Worklog Time Spent: 10m Work Description: jkff commented on a change in pull request #5141: [BEAM-3925] Allow ValueProvider for KafkaIO so that we can create Beam Templates using KafkaIO URL: https://github.com/apache/beam/pull/5141#discussion_r183140040 ########## File path: sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java ########## @@ -870,25 +1092,133 @@ private KafkaIO() {} * determine partition in Kafka (see {@link ProducerRecord} for more details). */ public Write<K, V> withKeySerializer(Class<? extends Serializer<K>> keySerializer) { - return toBuilder().setKeySerializer(keySerializer).build(); + return toBuilder() + .setKeySerializer(ValueProvider.StaticValueProvider.of(keySerializer)).build(); } /** - * Sets a {@link Serializer} for serializing value to bytes. + * Sets a {@link Serializer} for serializing key (if any) to bytes using a String. + */ + public Write<K, V> withKeySerializer(String keySerializer) { + return withKeySerializer(ValueProvider.StaticValueProvider.of(keySerializer)); + } + + /** + * Like above but with a {@link ValueProvider ValueProvider<String>}. + */ + public Write<K, V> withKeySerializer(ValueProvider<String> keySerializer) { + return toBuilder().setKeySerializer(ValueProvider. + NestedValueProvider.of(keySerializer, new SerializerKeyTranslator())).build(); + } + + /** + * Sets a {@link Serializer Serializer<V>} for serializing value to bytes. */ public Write<K, V> withValueSerializer(Class<? extends Serializer<V>> valueSerializer) { - return toBuilder().setValueSerializer(valueSerializer).build(); + return toBuilder().setValueSerializer( + ValueProvider.StaticValueProvider.of(valueSerializer)).build(); + } + + /** + * Like above but with a class name provided as a {@link String}. + */ + public Write<K, V> withValueSerializer(String valueSerializer) { + return withValueSerializer(ValueProvider.StaticValueProvider.of(valueSerializer)); + } + + /** + * Like above but with a {@link ValueProvider ValueProvider<String>}. + */ + public Write<K, V> withValueSerializer(ValueProvider<String> valueSerializer) { + return toBuilder().setValueSerializer(ValueProvider + .NestedValueProvider.of(valueSerializer, new SerializerValueTranslator())).build(); + } + + /** + * Used to build a {@link ValueProvider} for {@link Serializer Serializer<K>}. + */ + private class SerializerKeyTranslator + implements SerializableFunction<String, Class<? extends Serializer<K>>> { + @SuppressWarnings("unchecked") + @Override + public Class apply(String serializer) { Review comment: These are all also redundant with the deserializer code. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on 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 Issue Time Tracking ------------------- Worklog Id: (was: 93366) Time Spent: 1.5h (was: 1h 20m) > Allow ValueProvider for KafkaIO > ------------------------------- > > Key: BEAM-3925 > URL: https://issues.apache.org/jira/browse/BEAM-3925 > Project: Beam > Issue Type: Improvement > Components: sdk-java-core > Reporter: Sameer Abhyankar > Assignee: Pramod Upamanyu > Priority: Major > Time Spent: 1.5h > Remaining Estimate: 0h > > Add ValueProvider support for the various methods in KafkaIO. This would > allow us to use KafkaIO in reusable pipeline templates. -- This message was sent by Atlassian JIRA (v7.6.3#76005)