Copilot commented on code in PR #4252:
URL: https://github.com/apache/flink-cdc/pull/4252#discussion_r2882901162
##########
docs/content.zh/docs/core-concept/transform.md:
##########
@@ -395,6 +395,44 @@ pipeline:
注意这里的 `classpath` 必须是全限定名，并且对应的 `jar` 文件必须包含在 Flink `/lib` 文件夹中，或者通过
`flink-cdc.sh --jar` 选项传递。
+### UDF 配置选项
+
+你可以通过添加 `options` 块来向 UDF 传递额外的配置选项。这些选项可以在 `open` 方法中通过
`UserDefinedFunctionContext.configuration()` 获取：
+
+```yaml
+pipeline:
+ user-defined-function:
+ - name: query_redis
+ classpath: com.example.flink.cdc.udf.RedisQueryFunction
+ options:
+ hostname: localhost
+ port: "6379"
+ cache.enabled: "true"
+```
+
+在你的 UDF 实现中，可以这样访问这些配置选项：
+
+```java
+public class RedisQueryFunction implements UserDefinedFunction {
+ private String hostname;
+ private int port;
+
+ @Override
+ public void open(UserDefinedFunctionContext context) throws Exception {
+ hostname = context.configuration().get("hostname");
+ port = Integer.parseInt(context.configuration().get("port"));
Review Comment:
The documentation code example uses
`context.configuration().get("hostname")` and
`context.configuration().get("port")`, but `Configuration.get()` only accepts a
`ConfigOption<T>` parameter, not a plain `String`. This would cause a
compilation error if users copy this code.
The correct approach (as shown in the example UDF classes included in this
PR) is to define `ConfigOption` instances and use them:
```
ConfigOption<String> HOSTNAME =
ConfigOptions.key("hostname").stringType().noDefaultValue();
ConfigOption<Integer> PORT =
ConfigOptions.key("port").intType().defaultValue(6379);
```
Then access them via `context.configuration().get(HOSTNAME)` and
`context.configuration().get(PORT)`.
```suggestion
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
public class RedisQueryFunction implements UserDefinedFunction {
private static final ConfigOption<String> HOSTNAME =
ConfigOptions.key("hostname").stringType().noDefaultValue();
private static final ConfigOption<Integer> PORT =
ConfigOptions.key("port").intType().defaultValue(6379);
private String hostname;
private int port;
@Override
public void open(UserDefinedFunctionContext context) throws Exception {
hostname = context.configuration().get(HOSTNAME);
port = context.configuration().get(PORT);
```
##########
docs/content/docs/core-concept/transform.md:
##########
@@ -400,6 +400,43 @@ pipeline:
Notice that given classpath must be fully-qualified, and corresponding `jar`
files must be included in Flink `/lib` folder, or be passed with `flink-cdc.sh
--jar` option.
+### UDF Options
+
+You can pass extra options to UDFs by adding an `options` block. These options
will be available in the `open` method through
`UserDefinedFunctionContext.configuration()`:
+
+```yaml
+pipeline:
+ user-defined-function:
+ - name: query_redis
+ classpath: com.example.flink.cdc.udf.RedisQueryFunction
+ options:
+ hostname: localhost
+ port: "6379"
+ cache.enabled: "true"
+```
+
+And in your UDF implementation, you can access these options:
+
+```java
+public class RedisQueryFunction implements UserDefinedFunction {
+ private String hostname;
+ private int port;
+
+ @Override
+ public void open(UserDefinedFunctionContext context) throws Exception {
+ hostname = context.configuration().get("hostname");
+ port = Integer.parseInt(context.configuration().get("port"));
+ // Initialize your connection here...
+ }
+
Review Comment:
The documentation code example uses
`context.configuration().get("hostname")` and
`context.configuration().get("port")`, but `Configuration.get()` only accepts a
`ConfigOption<T>` parameter, not a plain `String`. This would cause a
compilation error if users copy this code.
The correct approach (as shown in the example UDF classes included in this
PR) is to define `ConfigOption` instances and use them:
```
ConfigOption<String> HOSTNAME =
ConfigOptions.key("hostname").stringType().noDefaultValue();
ConfigOption<Integer> PORT =
ConfigOptions.key("port").intType().defaultValue(6379);
```
Then access them via `context.configuration().get(HOSTNAME)` and
`context.configuration().get(PORT)`.
```suggestion
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
public class RedisQueryFunction implements UserDefinedFunction {
private static final ConfigOption<String> HOSTNAME =
ConfigOptions.key("hostname").stringType().noDefaultValue();
private static final ConfigOption<Integer> PORT =
ConfigOptions.key("port").intType().defaultValue(6379);
private String hostname;
private int port;
@Override
public void open(UserDefinedFunctionContext context) throws Exception {
hostname = context.configuration().get(HOSTNAME);
port = context.configuration().get(PORT);
// Initialize your connection here...
}
```
--
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]
