voonhous commented on code in PR #19760:
URL: https://github.com/apache/hudi/pull/19760#discussion_r3870157816
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java:
##########
@@ -311,6 +344,59 @@ void testConfigureSchemaDeserializer() throws IOException {
assertEquals(StringUtils.concatenateWithThreshold(String.format("%s_",
groupId), schemaHash, GROUP_ID_MAX_BYTES_LENGTH), newGroupId);
}
+ /**
+ * Covers the per-fetch re-configure branch of {@link
AvroKafkaSource#readFromCheckpoint}, which regressed twice
+ * (#10118 and #12111). When the source schema evolves between two fetches
on the same source instance, the second
+ * batch must decode with the refreshed schema instead of the one the source
was constructed with.
+ */
+ @Test
+ void testSchemaDeserializerRefreshesSchemaBetweenFetches() throws
IOException {
+ final String topic = TEST_TOPIC_PREFIX + "testSchemaDeserializerRefresh";
+ // a scope of its own keeps this test's registrations out of the registry
shared by the other mock:// tests
+ final String registryUrl = "mock://" + topic;
+ HoodieSchema simpleSchema =
loadSchemaFromResource("schema/simple-test-with-default-value.avsc");
+ HoodieSchema evolvedSchema =
loadSchemaFromResource("schema/evolved-test-with-default-value.avsc");
+
+ HoodieSchema previousSchema =
SchemaTestProvider.schemaToReturn.getAndSet(simpleSchema);
+ try {
+ // single partition so both fetches map to the same spark consumer-cache
key (group.id + partition),
+ // leaving the rotated group.id as the only thing separating the old
consumer from the new one
+ testUtils.createTopic(topic, 1);
+ TypedProperties props = createPropsForKafkaSource(topic, null,
"earliest");
+ props.put(KAFKA_AVRO_VALUE_DESERIALIZER_CLASS.key(),
KafkaAvroSchemaDeserializer.class.getName());
+ props.put("schema.registry.url", registryUrl);
+ AvroKafkaSource avroKafkaSource = new AvroKafkaSource(props, jsc(),
spark(), new SchemaTestProvider(props), metrics);
+
+ sendUserRecordsWithConfluentSerializer(topic, registryUrl, simpleSchema,
5, null);
+ InputBatch<JavaRDD<GenericRecord>> fetch1 =
avroKafkaSource.fetchNext(Option.empty(), Long.MAX_VALUE);
+ List<GenericRecord> firstBatch = fetch1.getBatch().get().collect();
+ assertEquals(5, firstBatch.size());
+ for (GenericRecord record : firstBatch) {
+ assertNull(record.getSchema().getField("phone"));
+ }
+ String groupIdAfterFirstFetch =
avroKafkaSource.props.getString(NATIVE_KAFKA_CONSUMER_GROUP_ID, "");
+
+ // evolve the source schema between fetches, as a continuous-mode
streamer would see it
+ SchemaTestProvider.schemaToReturn.set(evolvedSchema);
+ sendUserRecordsWithConfluentSerializer(topic, registryUrl,
evolvedSchema, 5, "555-0100");
+
+ InputBatch<JavaRDD<GenericRecord>> fetch2 =
avroKafkaSource.fetchNext(Option.of(fetch1.getCheckpointForNextBatch()),
Long.MAX_VALUE);
+ List<GenericRecord> secondBatch = fetch2.getBatch().get().collect();
+ assertEquals(5, secondBatch.size());
+ for (GenericRecord record : secondBatch) {
+ assertEquals(evolvedSchema.toAvroSchema(), record.getSchema());
Review Comment:
Added in 95aa4002f492: 3 of the 8 records in the second batch are written
under the simple schema after the provider has evolved; the test asserts all 8
come back with the evolved schema and that exactly those 3 have a `null` phone.
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java:
##########
@@ -311,6 +344,59 @@ void testConfigureSchemaDeserializer() throws IOException {
assertEquals(StringUtils.concatenateWithThreshold(String.format("%s_",
groupId), schemaHash, GROUP_ID_MAX_BYTES_LENGTH), newGroupId);
}
+ /**
+ * Covers the per-fetch re-configure branch of {@link
AvroKafkaSource#readFromCheckpoint}, which regressed twice
+ * (#10118 and #12111). When the source schema evolves between two fetches
on the same source instance, the second
+ * batch must decode with the refreshed schema instead of the one the source
was constructed with.
+ */
+ @Test
+ void testSchemaDeserializerRefreshesSchemaBetweenFetches() throws
IOException {
+ final String topic = TEST_TOPIC_PREFIX + "testSchemaDeserializerRefresh";
+ // a scope of its own keeps this test's registrations out of the registry
shared by the other mock:// tests
+ final String registryUrl = "mock://" + topic;
+ HoodieSchema simpleSchema =
loadSchemaFromResource("schema/simple-test-with-default-value.avsc");
+ HoodieSchema evolvedSchema =
loadSchemaFromResource("schema/evolved-test-with-default-value.avsc");
+
+ HoodieSchema previousSchema =
SchemaTestProvider.schemaToReturn.getAndSet(simpleSchema);
+ try {
+ // single partition so both fetches map to the same spark consumer-cache
key (group.id + partition),
+ // leaving the rotated group.id as the only thing separating the old
consumer from the new one
+ testUtils.createTopic(topic, 1);
+ TypedProperties props = createPropsForKafkaSource(topic, null,
"earliest");
+ props.put(KAFKA_AVRO_VALUE_DESERIALIZER_CLASS.key(),
KafkaAvroSchemaDeserializer.class.getName());
+ props.put("schema.registry.url", registryUrl);
+ AvroKafkaSource avroKafkaSource = new AvroKafkaSource(props, jsc(),
spark(), new SchemaTestProvider(props), metrics);
+
+ sendUserRecordsWithConfluentSerializer(topic, registryUrl, simpleSchema,
5, null);
+ InputBatch<JavaRDD<GenericRecord>> fetch1 =
avroKafkaSource.fetchNext(Option.empty(), Long.MAX_VALUE);
+ List<GenericRecord> firstBatch = fetch1.getBatch().get().collect();
+ assertEquals(5, firstBatch.size());
+ for (GenericRecord record : firstBatch) {
+ assertNull(record.getSchema().getField("phone"));
+ }
+ String groupIdAfterFirstFetch =
avroKafkaSource.props.getString(NATIVE_KAFKA_CONSUMER_GROUP_ID, "");
+
+ // evolve the source schema between fetches, as a continuous-mode
streamer would see it
+ SchemaTestProvider.schemaToReturn.set(evolvedSchema);
+ sendUserRecordsWithConfluentSerializer(topic, registryUrl,
evolvedSchema, 5, "555-0100");
+
+ InputBatch<JavaRDD<GenericRecord>> fetch2 =
avroKafkaSource.fetchNext(Option.of(fetch1.getCheckpointForNextBatch()),
Long.MAX_VALUE);
+ List<GenericRecord> secondBatch = fetch2.getBatch().get().collect();
+ assertEquals(5, secondBatch.size());
+ for (GenericRecord record : secondBatch) {
+ assertEquals(evolvedSchema.toAvroSchema(), record.getSchema());
+ assertEquals("555-0100", record.get("phone").toString());
+ }
+
+ // pin the mechanism both past fixes introduced: the stamped reader
schema and the group.id rotation
+ assertEquals(evolvedSchema.toString(),
avroKafkaSource.props.getString(KAFKA_VALUE_DESERIALIZER_SCHEMA.key()));
+ assertNotEquals(groupIdAfterFirstFetch,
avroKafkaSource.props.getString(NATIVE_KAFKA_CONSUMER_GROUP_ID, ""));
Review Comment:
Now asserting the exact `concatenateWithThreshold(groupIdAfterFirstFetch +
"_", hash(evolvedSchema), GROUP_ID_MAX_BYTES_LENGTH)` value in 95aa4002f492.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
