[GitHub] [beam] kennknowles commented on a diff in pull request #21783: Eliminate nullness errors in KafkaIO

[GitBox <gi...@apache.org>]

kennknowles commented on code in PR #21783:
URL: https://github.com/apache/beam/pull/21783#discussion_r894782013


##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java:
##########
@@ -117,7 +118,7 @@ public static void evaluateSeek2End(Consumer<?, ?> consumer, TopicPartition topi
   }
 
   public static void evaluateAssign(
-      Consumer<?, ?> consumer, Collection<TopicPartition> topicPartitions) {
+      @Nullable Consumer<?, ?> consumer, @Nullable Collection<TopicPartition> topicPartitions) {

Review Comment:
   Done



##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java:
##########
@@ -691,33 +700,36 @@ void insert(int shard, ShardWriter<K, V> writer) {
    * partitions for a topic rather than for fetching messages.
    */
   private static Consumer<?, ?> openConsumer(WriteRecords<?, ?> spec) {
-    return spec.getConsumerFactoryFn()
-        .apply(
-            ImmutableMap.of(
-                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
-                spec.getProducerConfig().get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
-                ConsumerConfig.GROUP_ID_CONFIG,
-                spec.getSinkGroupId(),
-                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
-                ByteArrayDeserializer.class,
-                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
-                ByteArrayDeserializer.class));
+    SerializableFunction<Map<String, Object>, ? extends Consumer<?, ?>> consumerFactoryFn =
+        Preconditions.checkArgumentNotNull(spec.getConsumerFactoryFn());
+
+    Map<String, Object> consumerConfig = new HashMap<>();
+    Object bootStrapServersConfig =

Review Comment:
   Done



-- 
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.

To unsubscribe, e-mail: github-unsubscribe@beam.apache.org

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