rename field

Author: stankiewicz

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java

@@ -162,8 +162,8 @@ static void ensureEOSSupport() {
   @Override
   public PCollection<Void> expand(PCollection<ProducerRecord<K, V>> input) {
     String topic = Preconditions.checkStateNotNull(spec.getTopic());
-    int numElements = spec.getNumElements();
-    Duration timeout = spec.getTimeout();
+    int numElements = spec.getEosTriggerNumElements();
+    Duration timeout = spec.getEosTriggerTimeout();
     int numShards = spec.getNumShards();
     if (numShards <= 0) {
       try (Consumer<?, ?> consumer = openConsumer(spec)) {

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -648,8 +648,8 @@ public static <K, V> WriteRecords<K, V> writeRecords() {
     return new AutoValue_KafkaIO_WriteRecords.Builder<K, V>()
         .setProducerConfig(WriteRecords.DEFAULT_PRODUCER_PROPERTIES)
         .setEOS(false)
-        .setNumElements(100)
-        .setTimeout(Duration.standardSeconds(1))
+        .setEosTriggerNumElements(100)
+        .setEosTriggerTimeout(Duration.standardSeconds(1))
         .setNumShards(0)
         .setConsumerFactoryFn(KafkaIOUtils.KAFKA_CONSUMER_FACTORY_FN)
         .setBadRecordRouter(BadRecordRouter.THROWING_ROUTER)
@@ -3187,9 +3187,9 @@ public abstract static class WriteRecords<K, V>
     @Pure
     public abstract boolean isEOS();
 
-    public abstract int getNumElements();
+    public abstract int getEosTriggerNumElements();
 
-    public abstract Duration getTimeout();
+    public abstract Duration getEosTriggerTimeout();
 
     @Pure
     public abstract @Nullable String getSinkGroupId();
@@ -3227,9 +3227,9 @@ abstract Builder<K, V> setPublishTimestampFunction(
 
       abstract Builder<K, V> setEOS(boolean eosEnabled);
 
-      abstract Builder<K, V> setNumElements(int numElements);
+      abstract Builder<K, V> setEosTriggerNumElements(int numElements);
 
-      abstract Builder<K, V> setTimeout(Duration timeout);
+      abstract Builder<K, V> setEosTriggerTimeout(Duration timeout);
 
       abstract Builder<K, V> setSinkGroupId(String sinkGroupId);
 
@@ -3381,7 +3381,7 @@ public WriteRecords<K, V> withEOS(int numShards, String sinkGroupId) {
     public WriteRecords<K, V> withEOSTriggerConfig(int numElements, Duration timeout) {
       checkArgument(numElements >= 1, "numElements should be >= 1");
       checkArgument(timeout != null, "timeout is required for exactly-once sink");
-      return toBuilder().setNumElements(numElements).setTimeout(timeout).build();
+      return toBuilder().setEosTriggerNumElements(numElements).setEosTriggerTimeout(timeout).build();
     }
 
     /**

sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java

@@ -479,6 +479,8 @@ static class KafkaIOWriteTranslator implements TransformPayloadTranslator<Write<
             .addNullableByteArrayField("producer_factory_fn")
             .addNullableByteArrayField("publish_timestamp_fn")
             .addBooleanField("eos")
+            .addInt32Field("eos_trigger_num_elements")
+            .addInt64Field("eos_trigger_timeout_ms")
             .addInt32Field("num_shards")
             .addNullableStringField("sink_group_id")
             .addNullableByteArrayField("consumer_factory_fn")
@@ -547,6 +549,9 @@ public Row toConfigRow(Write<?, ?> transform) {
       }
 
       fieldValues.put("eos", writeRecordsTransform.isEOS());
+      fieldValues.put(
+          "eos_trigger_timeout_ms", writeRecordsTransform.getEosTriggerTimeout().getMillis());
+      fieldValues.put("eos_trigger_num_elements", writeRecordsTransform.getEosTriggerNumElements());
       fieldValues.put("num_shards", writeRecordsTransform.getNumShards());
 
       if (writeRecordsTransform.getSinkGroupId() != null) {

sdks/java/io/kafka/upgrade/src/test/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslationTest.java

@@ -94,6 +94,8 @@ public class KafkaIOTranslationTest {
     WRITE_TRANSFORM_SCHEMA_MAPPING.put("getValueSerializer", "value_serializer");
     WRITE_TRANSFORM_SCHEMA_MAPPING.put("getPublishTimestampFunction", "publish_timestamp_fn");
     WRITE_TRANSFORM_SCHEMA_MAPPING.put("isEOS", "eos");
+    WRITE_TRANSFORM_SCHEMA_MAPPING.put("getEosTriggerTimeout", "eos_trigger_timeout_ms");
+    WRITE_TRANSFORM_SCHEMA_MAPPING.put("getEosTriggerNumElements", "eos_trigger_num_elements");
     WRITE_TRANSFORM_SCHEMA_MAPPING.put("getSinkGroupId", "sink_group_id");
     WRITE_TRANSFORM_SCHEMA_MAPPING.put("getNumShards", "num_shards");
     WRITE_TRANSFORM_SCHEMA_MAPPING.put("getConsumerFactoryFn", "consumer_factory_fn");