added missing schemaFieldNumber annotation

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

@@ -174,22 +174,27 @@ public static Builder builder() {
   public abstract ErrorHandling getErrorHandling();
 
   @SchemaFieldDescription("If the Kafka read should be redistributed.")
+  @SchemaFieldNumber("12")
   @Nullable
   public abstract Boolean getRedistributed();
 
   @SchemaFieldDescription("If the Kafka read allows duplicates.")
+  @SchemaFieldNumber("13")
   @Nullable
   public abstract Boolean getAllowDuplicates();
 
   @SchemaFieldDescription("The number of keys for redistributing Kafka inputs.")
+  @SchemaFieldNumber("14")
   @Nullable
   public abstract Integer getRedistributeNumKeys();
 
   @SchemaFieldDescription("If the redistribute is using offset deduplication mode.")
+  @SchemaFieldNumber("15")
   @Nullable
   public abstract Boolean getOffsetDeduplication();
 
   @SchemaFieldDescription("If the redistribute keys by the Kafka record key.")
+  @SchemaFieldNumber("16")
   @Nullable
   public abstract Boolean getRedistributeByRecordKey();
 

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

@@ -380,7 +380,7 @@ public void testKafkaReadSchemaTransformConfigurationSchema() throws NoSuchSchem
     Schema schema =
         SchemaRegistry.createDefault().getSchema(KafkaReadSchemaTransformConfiguration.class);
 
-    assertEquals(12, schema.getFieldCount());
+    assertEquals(17, schema.getFieldCount());
 
     // Check field name, type, and nullability. Descriptions are not checked as they are not
     // critical for serialization.
@@ -453,5 +453,30 @@ public void testKafkaReadSchemaTransformConfigurationSchema() throws NoSuchSchem
                                 actualRowSchemaForErrorHandling.getField(0).getDescription()))))
             .withDescription(schema.getField(11).getDescription()),
         schema.getField(11));
+
+    assertEquals(
+        Schema.Field.nullable("redistributed", Schema.FieldType.BOOLEAN)
+            .withDescription(schema.getField(12).getDescription()),
+        schema.getField(12));
+
+    assertEquals(
+        Schema.Field.nullable("allowDuplicates", Schema.FieldType.BOOLEAN)
+            .withDescription(schema.getField(13).getDescription()),
+        schema.getField(13));
+
+    assertEquals(
+        Schema.Field.nullable("redistributeNumKeys", Schema.FieldType.INT32)
+            .withDescription(schema.getField(14).getDescription()),
+        schema.getField(14));
+
+    assertEquals(
+        Schema.Field.nullable("offsetDeduplication", Schema.FieldType.BOOLEAN)
+            .withDescription(schema.getField(15).getDescription()),
+        schema.getField(15));
+
+    assertEquals(
+        Schema.Field.nullable("redistributeByRecordKey", Schema.FieldType.BOOLEAN)
+            .withDescription(schema.getField(16).getDescription()),
+        schema.getField(16));
   }
 }