Abacn commented on code in PR #35085:
URL: https://github.com/apache/beam/pull/35085#discussion_r2162024328
##########
sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaReadSchemaTransformProvider.java:
##########
@@ -178,16 +180,45 @@ public PCollectionRowTuple expand(PCollectionRowTuple
input) {
if (confluentSchemaRegUrl != null) {
final String confluentSchemaRegSubject =
checkArgumentNotNull(configuration.getConfluentSchemaRegistrySubject());
- KafkaIO.Read<byte[], GenericRecord> kafkaRead =
- KafkaIO.<byte[], GenericRecord>read()
- .withTopic(configuration.getTopic())
- .withConsumerFactoryFn(new ConsumerFactoryWithGcsTrustStores())
- .withBootstrapServers(configuration.getBootstrapServers())
- .withConsumerConfigUpdates(consumerConfigs)
- .withKeyDeserializer(ByteArrayDeserializer.class)
- .withValueDeserializer(
- ConfluentSchemaRegistryDeserializerProvider.of(
- confluentSchemaRegUrl, confluentSchemaRegSubject));
+ KafkaIO.Read<byte[], GenericRecord> kafkaRead;
+
+ if (confluentSchemaRegUrl.contains(googleManagedSchemaRegistryPrefix))
{
Review Comment:
For readability we may want to generalize this logic. For example, put
common with... out of if clause, and create a helper method to handle certain
confluentSchemaRegUrl. This makes it easy to extend in the future
--
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: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
