Skip to content

Commit 090b17b

Browse files
authored
added missing schemaFieldNumber annotation (#36489)
1 parent 2a77c59 commit 090b17b

2 files changed

Lines changed: 31 additions & 1 deletion

File tree

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,22 +174,27 @@ public static Builder builder() {
174174
public abstract ErrorHandling getErrorHandling();
175175

176176
@SchemaFieldDescription("If the Kafka read should be redistributed.")
177+
@SchemaFieldNumber("12")
177178
@Nullable
178179
public abstract Boolean getRedistributed();
179180

180181
@SchemaFieldDescription("If the Kafka read allows duplicates.")
182+
@SchemaFieldNumber("13")
181183
@Nullable
182184
public abstract Boolean getAllowDuplicates();
183185

184186
@SchemaFieldDescription("The number of keys for redistributing Kafka inputs.")
187+
@SchemaFieldNumber("14")
185188
@Nullable
186189
public abstract Integer getRedistributeNumKeys();
187190

188191
@SchemaFieldDescription("If the redistribute is using offset deduplication mode.")
192+
@SchemaFieldNumber("15")
189193
@Nullable
190194
public abstract Boolean getOffsetDeduplication();
191195

192196
@SchemaFieldDescription("If the redistribute keys by the Kafka record key.")
197+
@SchemaFieldNumber("16")
193198
@Nullable
194199
public abstract Boolean getRedistributeByRecordKey();
195200

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

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -380,7 +380,7 @@ public void testKafkaReadSchemaTransformConfigurationSchema() throws NoSuchSchem
380380
Schema schema =
381381
SchemaRegistry.createDefault().getSchema(KafkaReadSchemaTransformConfiguration.class);
382382

383-
assertEquals(12, schema.getFieldCount());
383+
assertEquals(17, schema.getFieldCount());
384384

385385
// Check field name, type, and nullability. Descriptions are not checked as they are not
386386
// critical for serialization.
@@ -453,5 +453,30 @@ public void testKafkaReadSchemaTransformConfigurationSchema() throws NoSuchSchem
453453
actualRowSchemaForErrorHandling.getField(0).getDescription()))))
454454
.withDescription(schema.getField(11).getDescription()),
455455
schema.getField(11));
456+
457+
assertEquals(
458+
Schema.Field.nullable("redistributed", Schema.FieldType.BOOLEAN)
459+
.withDescription(schema.getField(12).getDescription()),
460+
schema.getField(12));
461+
462+
assertEquals(
463+
Schema.Field.nullable("allowDuplicates", Schema.FieldType.BOOLEAN)
464+
.withDescription(schema.getField(13).getDescription()),
465+
schema.getField(13));
466+
467+
assertEquals(
468+
Schema.Field.nullable("redistributeNumKeys", Schema.FieldType.INT32)
469+
.withDescription(schema.getField(14).getDescription()),
470+
schema.getField(14));
471+
472+
assertEquals(
473+
Schema.Field.nullable("offsetDeduplication", Schema.FieldType.BOOLEAN)
474+
.withDescription(schema.getField(15).getDescription()),
475+
schema.getField(15));
476+
477+
assertEquals(
478+
Schema.Field.nullable("redistributeByRecordKey", Schema.FieldType.BOOLEAN)
479+
.withDescription(schema.getField(16).getDescription()),
480+
schema.getField(16));
456481
}
457482
}

0 commit comments

Comments
 (0)