[TUBEMQ-441]An error occurred when using the Tubemq class to create a…

… sink table (#339)

Co-authored-by: gosonzhang <gosonzhang@tencent.com>

tubemq-connectors/tubemq-connector-flink/src/main/java/org/apache/flink/connectors/tubemq/Tubemq.java

@@ -40,6 +40,9 @@
  */
 public class Tubemq extends ConnectorDescriptor {
 
+    @Nullable
+    private boolean consumerRole = true;
+
     @Nullable
     private String topic;
 
@@ -73,6 +76,16 @@ public Tubemq topic(String topic) {
         return this;
     }
 
+    /**
+     * Sets the client role to be used.
+     *
+     * @param isConsumer The client role if consumer.
+     */
+    public Tubemq asConsumer(boolean isConsumer) {
+        this.consumerRole = isConsumer;
+        return this;
+    }
+
     /**
      * Sets the address of tubemq master to connect.
      *
@@ -133,13 +146,14 @@ protected Map<String, String> toConnectorProperties() {
         if (master != null) {
             descriptorProperties.putString(CONNECTOR_MASTER, master);
         }
-
-        if (group != null) {
-            descriptorProperties.putString(CONNECTOR_GROUP, group);
-        }
-
-        if (tids != null) {
-            descriptorProperties.putString(CONNECTOR_TIDS, tids);
+        if (consumerRole) {
+            if (group != null) {
+                descriptorProperties.putString(CONNECTOR_GROUP, group);
+            }
+
+            if (tids != null) {
+                descriptorProperties.putString(CONNECTOR_TIDS, tids);
+            }
         }
 
         descriptorProperties.putPropertiesWithPrefix(CONNECTOR_PROPERTIES, properties);