nsivabalan commented on code in PR #13604:
URL: https://github.com/apache/hudi/pull/13604#discussion_r2227087747
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaSourceUtil.java:
##########
@@ -47,4 +49,36 @@ public static void
configureSchemaDeserializer(SchemaProvider schemaProvider, Ty
: StringUtils.concatenateWithThreshold(String.format("%s_", groupId),
schemaHash, GROUP_ID_MAX_BYTES_LENGTH);
props.put(NATIVE_KAFKA_CONSUMER_GROUP_ID, updatedConsumerGroup);
}
+
+ /**
+ * Utility method that removes configs with keys that match (start with) any
one of the prefixes.
+ *
+ * @param kafkaParams The incoming kafka params
+ * @param commaSeparatedPrefixes all configs with keys starting with any one
of these comma-separated prefixes will be ignored.
+ * @return a new set of kafkaParams with the configs matching the prefixes
removed.
+ */
+ public static Map<String, Object> filterKafkaParameters(Map<String, Object>
kafkaParams, String commaSeparatedPrefixes) {
+ if (commaSeparatedPrefixes.isEmpty()) {
+ return kafkaParams;
+ }
+
+ String[] prefixes = commaSeparatedPrefixes.split(";");
+ Map<String, Object> filteredInParams = new HashMap<>();
+ for (Map.Entry<String, Object> entry : kafkaParams.entrySet()) {
+ boolean beginsWithAtleastOnePrefix = false;
+ for (String prefix : prefixes) {
+ if (!prefix.isEmpty() && entry.getKey().startsWith(prefix)) {
Review Comment:
we can trim empty prefixes at L65 only and avoid first condition here
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaSourceUtil.java:
##########
@@ -47,4 +49,36 @@ public static void
configureSchemaDeserializer(SchemaProvider schemaProvider, Ty
: StringUtils.concatenateWithThreshold(String.format("%s_", groupId),
schemaHash, GROUP_ID_MAX_BYTES_LENGTH);
props.put(NATIVE_KAFKA_CONSUMER_GROUP_ID, updatedConsumerGroup);
}
+
+ /**
+ * Utility method that removes configs with keys that match (start with) any
one of the prefixes.
+ *
+ * @param kafkaParams The incoming kafka params
+ * @param commaSeparatedPrefixes all configs with keys starting with any one
of these comma-separated prefixes will be ignored.
+ * @return a new set of kafkaParams with the configs matching the prefixes
removed.
+ */
+ public static Map<String, Object> filterKafkaParameters(Map<String, Object>
kafkaParams, String commaSeparatedPrefixes) {
+ if (commaSeparatedPrefixes.isEmpty()) {
+ return kafkaParams;
+ }
+
+ String[] prefixes = commaSeparatedPrefixes.split(";");
Review Comment:
documentation says, comma separated list of prefixes and here we are
splitting by `;` ?
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/KafkaSourceUtil.java:
##########
@@ -47,4 +49,36 @@ public static void
configureSchemaDeserializer(SchemaProvider schemaProvider, Ty
: StringUtils.concatenateWithThreshold(String.format("%s_", groupId),
schemaHash, GROUP_ID_MAX_BYTES_LENGTH);
props.put(NATIVE_KAFKA_CONSUMER_GROUP_ID, updatedConsumerGroup);
}
+
+ /**
+ * Utility method that removes configs with keys that match (start with) any
one of the prefixes.
+ *
+ * @param kafkaParams The incoming kafka params
+ * @param commaSeparatedPrefixes all configs with keys starting with any one
of these comma-separated prefixes will be ignored.
+ * @return a new set of kafkaParams with the configs matching the prefixes
removed.
+ */
+ public static Map<String, Object> filterKafkaParameters(Map<String, Object>
kafkaParams, String commaSeparatedPrefixes) {
Review Comment:
does this need to be public?
##########
hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/helpers/TestKafkaSourceUtil.java:
##########
@@ -63,4 +66,23 @@ void testConfigureSchemaDeserializer() {
String schemaHash = Base64.encode(HashID.hash(avroSchemaJson,
HashID.Size.BITS_128));
assertEquals(props.getString(NATIVE_KAFKA_CONSUMER_GROUP_ID, ""),
schemaHash);
}
+
+ @Test
+ public void testFilterKafkaParameters() {
+ Map<String, Object> kafkaParams = new HashMap<>();
+
+ kafkaParams.put("custom1.config.streamer", "offer");
+ kafkaParams.put("boostrap.servers", "dns:port");
+ kafkaParams.put("custom2.config.capture", "s3://folder1");
+ kafkaParams.put("custom1config.sourceprofile.refresh.mode", "ENABLED");
+ // Case 1: No prefixes are configured.
+ assertEquals(kafkaParams,
KafkaSourceUtil.filterKafkaParameters(kafkaParams, ""));
+ // Case 2: Ensure only the appropriate configs are filtered out.
+ Map<String, Object> filteredParams =
KafkaSourceUtil.filterKafkaParameters(kafkaParams, "custom1,custom2");
+ Map<String, Object> expectedParams = new HashMap<>();
+ expectedParams.put("boostrap.servers", "dns:port");
+ assertEquals(expectedParams, filteredParams);
Review Comment:
just 1 additional ask.
can you also test "config2." to ensure its not regex matching, but exact
prefix matching.
--
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]
