You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ni...@apache.org on 2022/09/29 11:54:21 UTC
[rocketmq-flink] branch main updated: Fix some problems with the catalog (#58)
This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-flink.git
The following commit(s) were added to refs/heads/main by this push:
new 6909cbc Fix some problems with the catalog (#58)
6909cbc is described below
commit 6909cbc5d2c3c6a0590d5a7bd32bd9f45cf18f97
Author: 李晓双 Li Xiao Shuang <64...@qq.com>
AuthorDate: Thu Sep 29 19:54:16 2022 +0800
Fix some problems with the catalog (#58)
---
.../org/apache/rocketmq/flink/catalog/RocketMQCatalogFactory.java | 2 ++
.../rocketmq/flink/catalog/RocketMQCatalogFactoryOptions.java | 2 +-
.../apache/rocketmq/flink/sink/table/RocketMQDynamicTableSink.java | 6 ++++--
3 files changed, 7 insertions(+), 3 deletions(-)
diff --git a/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactory.java b/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactory.java
index 37d0b3c..609708f 100644
--- a/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactory.java
+++ b/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactory.java
@@ -61,6 +61,8 @@ public class RocketMQCatalogFactory implements CatalogFactory {
public Set<ConfigOption<?>> optionalOptions() {
Set<ConfigOption<?>> options = new HashSet<>();
options.add(DEFAULT_DATABASE);
+ options.add(NAME_SERVER_ADDR);
+ options.add(SCHEMA_REGISTRY_BASE_URL);
return options;
}
}
diff --git a/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactoryOptions.java b/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactoryOptions.java
index 25226b2..624539b 100644
--- a/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactoryOptions.java
+++ b/src/main/java/org/apache/rocketmq/flink/catalog/RocketMQCatalogFactoryOptions.java
@@ -30,7 +30,7 @@ import org.apache.flink.table.catalog.CommonCatalogOptions;
@Internal
public final class RocketMQCatalogFactoryOptions {
- public static final String IDENTIFIER = "rocketmq-catalog";
+ public static final String IDENTIFIER = "rocketmq_catalog";
public static final ConfigOption<String> DEFAULT_DATABASE =
ConfigOptions.key(CommonCatalogOptions.DEFAULT_DATABASE_KEY)
diff --git a/src/main/java/org/apache/rocketmq/flink/sink/table/RocketMQDynamicTableSink.java b/src/main/java/org/apache/rocketmq/flink/sink/table/RocketMQDynamicTableSink.java
index f61cbda..366991d 100644
--- a/src/main/java/org/apache/rocketmq/flink/sink/table/RocketMQDynamicTableSink.java
+++ b/src/main/java/org/apache/rocketmq/flink/sink/table/RocketMQDynamicTableSink.java
@@ -242,8 +242,10 @@ public class RocketMQDynamicTableSink implements DynamicTableSink, SupportsWriti
Properties producerProps = new Properties();
producerProps.setProperty(RocketMQConfig.PRODUCER_GROUP, producerGroup);
producerProps.setProperty(RocketMQConfig.NAME_SERVER_ADDR, nameServerAddress);
- producerProps.setProperty(RocketMQConfig.ACCESS_KEY, accessKey);
- producerProps.setProperty(RocketMQConfig.SECRET_KEY, secretKey);
+ if (accessKey != null && secretKey != null) {
+ producerProps.setProperty(RocketMQConfig.ACCESS_KEY, accessKey);
+ producerProps.setProperty(RocketMQConfig.SECRET_KEY, secretKey);
+ }
return producerProps;
}