FANNG1 commented on code in PR #6303:
URL: https://github.com/apache/gravitino/pull/6303#discussion_r1923147199
##########
flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java:
##########
@@ -32,23 +34,76 @@ public interface PropertiesConverter {
/**
* Converts properties from application provided properties and Flink
connector properties to
- * Gravitino properties.
+ * Gravitino properties.This method processes the Flink configuration and
transforms it into a
+ * format suitable for the Gravitino catalog.
*
- * @param flinkConf The configuration provided by Flink.
- * @return properties for the Gravitino catalog.
+ * @param flinkConf The Flink configuration containing connector properties.
This includes both
+ * Flink-specific properties and any user-provided properties.
+ * @return A map of properties converted for use in the Gravitino catalog.
The returned map
+ * includes both directly transformed properties and bypass properties
prefixed with {@link
+ * #FLINK_PROPERTY_PREFIX}.
*/
default Map<String, String> toGravitinoCatalogProperties(Configuration
flinkConf) {
- return flinkConf.toMap();
+ Map<String, String> gravitinoProperties = Maps.newHashMap();
+ for (Map.Entry<String, String> entry : flinkConf.toMap().entrySet()) {
+ String gravitinoKey =
transformPropertyToGravitinoCatalog(entry.getKey());
+ if (gravitinoKey != null) {
+ gravitinoProperties.put(gravitinoKey, entry.getValue());
+ } else if (!entry.getKey().startsWith(FLINK_PROPERTY_PREFIX)) {
+ gravitinoProperties.put(FLINK_PROPERTY_PREFIX + entry.getKey(),
entry.getValue());
+ } else {
+ gravitinoProperties.put(entry.getKey(), entry.getValue());
+ }
+ }
+ return gravitinoProperties;
}
/**
- * Converts properties from Gravitino properties to Flink connector
properties.
+ * Converts properties from Gravitino catalog properties to Flink connector
properties. This
+ * method processes the Gravitino properties and transforms them into a
format suitable for the
+ * Flink connector.
*
- * @param gravitinoProperties The properties provided by Gravitino.
- * @return properties for the Flink connector.
+ * @param gravitinoProperties The properties provided by the Gravitino
catalog. This includes both
+ * Gravitino-specific properties and any bypass properties prefixed with
{@link
+ * #FLINK_PROPERTY_PREFIX}.
+ * @return A map of properties converted for use in the Flink connector. The
returned map includes
+ * both transformed properties and the Flink catalog type.
*/
default Map<String, String> toFlinkCatalogProperties(Map<String, String>
gravitinoProperties) {
- return gravitinoProperties;
+ Map<String, String> all = Maps.newHashMap();
+ gravitinoProperties.forEach(
+ (key, value) -> {
+ String flinkConfigKey = key;
+ if (key.startsWith(PropertiesConverter.FLINK_PROPERTY_PREFIX)) {
+ flinkConfigKey =
key.substring(PropertiesConverter.FLINK_PROPERTY_PREFIX.length());
+ }
+ all.put(flinkConfigKey, value);
+ });
+ Map<String, String> allProperties = transformPropertiesToFlinkCatalog(all);
+ allProperties.put(CommonCatalogOptions.CATALOG_TYPE.key(),
getFlinkCatalogType());
+ return allProperties;
+ }
+
+ /**
+ * Transforms a Flink configuration key to a corresponding Gravitino catalog
property key. This
+ * method is used to map Flink-specific configuration keys to Gravitino
catalog properties.
+ *
+ * @param configKey The Flink configuration key to be transformed.
+ * @return The corresponding Gravitino catalog property key, or {@code null}
if no transformation
+ * is needed.
+ */
+ String transformPropertyToGravitinoCatalog(String configKey);
+
+ /**
+ * Transforms a map of properties from Gravitino catalog properties to Flink
connector properties.
+ * This method is used to convert properties that are specific to Gravitino
into a format that can
+ * be understood by the Flink connector.
+ *
+ * @param allProperties A map of properties to be transformed.
+ * @return A map of properties transformed into Flink connector properties.
+ */
+ default Map<String, String> transformPropertiesToFlinkCatalog(Map<String,
String> allProperties) {
Review Comment:
could we pass single gravitino property instead all properties as method
argument?
```
String transformPropertyToFlinkCatalog(String gravitinoProperty);
```
--
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]
