wombatu-kun commented on code in PR #19760:
URL: https://github.com/apache/hudi/pull/19760#discussion_r3868376715
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestAvroKafkaSource.java:
##########
@@ -151,6 +159,31 @@ void sendMessagesToKafkaWithNullKafkaValue(String topic,
int count, int numParti
}
}
+ private void sendUserRecordsWithConfluentSerializer(String topic, String
registryUrl, HoodieSchema schema, int count, String phone) {
+ Properties config = getProducerProperties();
+ config.put("value.serializer", KafkaAvroSerializer.class.getName());
+ config.put("schema.registry.url", registryUrl);
+ config.put("auto.register.schemas", "true");
+ try (Producer<String, GenericRecord> producer = new
KafkaProducer<>(config)) {
+ for (int i = 0; i < count; i++) {
+ GenericRecordBuilder builder = new
GenericRecordBuilder(schema.toAvroSchema())
+ .set("name", "user" + i)
+ .set("favorite_number", i)
+ .set("favorite_color", "blue");
+ if (phone != null) {
+ builder.set("phone", phone);
+ }
+ producer.send(new ProducerRecord<>(topic, "key", builder.build()));
+ }
+ }
+ }
+
+ private HoodieSchema loadSchemaFromResource(String resourcePath) throws
IOException {
Review Comment:
`SchemaTestUtil.getSchemaFromResource` already returns a `HoodieSchema` from
a test resource and is on this module's test classpath -
`TestGcsEventsHoodieIncrSource` and `S3EventsHoodieIncrSourceHarness` both call
it. Using it here (leading `/` on the path, since it resolves via
`Class.getResourceAsStream`) also lets the test method drop `throws
IOException`.
##########
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:
Both fetches pair the writer schema with an identical reader schema, so no
assertion here depends on the injected `sourceSchema` actually being used - the
old-record-under-evolved-schema case `TestKafkaAvroSchemaDeserializer` covers
never happens on the source path. Producing part of the second batch under
`simpleSchema` and asserting those come back with `evolvedSchema` and a null
`phone` would cover it - intentional, or worth adding?
##########
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
Review Comment:
Neither PR fixed a regression - before #10118 the schema was stamped only in
the constructor, so #10118 introduced this per-fetch re-configure and #12111
added the `group.id` rotation on top of it. Naming the two mechanisms instead
of "regressed twice" would line each one up with the assertion below that pins
it.
##########
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:
`KafkaSourceUtil.configureSchemaDeserializer` appends the schema hash to
`group.id` on every call, not only when the schema changed, so this holds for
any two fetches. Asserting the exact value the way
`testConfigureSchemaDeserializer` does -
`concatenateWithThreshold(groupIdAfterFirstFetch + "_",
Base64.encode(HashID.hash(evolvedSchema.toString(), HashID.Size.BITS_128)),
GROUP_ID_MAX_BYTES_LENGTH)` - would pin the rotation to the evolved schema.
--
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]
