[improve][connector] Add getSourceConfig method on SourceContext (ap…

…ache#16305)
bestming8 · Jul 13, 2022 · c122928 · c122928
1 parent 840e920
commit c122928
Show file tree

Hide file tree

Showing 4 changed files with 28 additions and 0 deletions.
diff --git a/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
@@ -57,6 +57,7 @@
 import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
 import org.apache.pulsar.client.impl.ProducerBuilderImpl;
 import org.apache.pulsar.common.io.SinkConfig;
+import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.pulsar.functions.api.Context;
@@ -76,6 +77,7 @@
 import org.apache.pulsar.functions.source.TopicSchema;
 import org.apache.pulsar.functions.utils.FunctionCommon;
 import org.apache.pulsar.functions.utils.SinkConfigUtils;
+import org.apache.pulsar.functions.utils.SourceConfigUtils;
 import org.apache.pulsar.io.core.SinkContext;
 import org.apache.pulsar.io.core.SourceContext;
 import org.slf4j.Logger;
@@ -264,6 +266,11 @@ public String getOutputTopic() {
         return config.getFunctionDetails().getSink().getTopic();
     }
 
+    @Override
+    public SourceConfig getSourceConfig() {
+        return SourceConfigUtils.convertFromDetails(config.getFunctionDetails());
+    }
+
     @Override
     public String getOutputSchemaType() {
         SinkSpec sink = config.getFunctionDetails().getSink();

diff --git a/...unctions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java b/...unctions/instance/src/test/java/org/apache/pulsar/functions/instance/ContextImplTest.java
@@ -150,6 +150,13 @@ public void testGetSinkConfig() {
         Assert.assertNotNull(sinkConfig);
     }
 
+    @Test
+    public void testGetSourceConfig() {
+        SinkContext sourceContext = context;
+        SinkConfig sinkConfig = sourceContext.getSinkConfig();
+        Assert.assertNotNull(sinkConfig);
+    }
+
     @Test
     public void testIncrCounterStateEnabled() throws Exception {
         context.defaultStateStore = mock(BKStateStoreImpl.class);

diff --git a/pulsar-io/common/src/test/java/org/apache/pulsar/io/common/IOConfigUtilsTest.java b/pulsar-io/common/src/test/java/org/apache/pulsar/io/common/IOConfigUtilsTest.java
@@ -26,6 +26,7 @@
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.common.io.SinkConfig;
+import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.io.core.SinkContext;
 import org.apache.pulsar.io.core.SourceContext;
 import org.apache.pulsar.io.core.annotations.FieldDoc;
@@ -123,6 +124,11 @@ public String getOutputTopic() {
             return null;
         }
 
+        @Override
+        public SourceConfig getSourceConfig() {
+            return null;
+        }
+
         @Override
         public String getTenant() {
             return null;

diff --git a/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/SourceContext.java b/pulsar-io/core/src/main/java/org/apache/pulsar/io/core/SourceContext.java
@@ -24,6 +24,7 @@
 import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
+import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.functions.api.BaseContext;
 
 /**
@@ -47,6 +48,13 @@ public interface SourceContext extends BaseContext {
      */
     String getOutputTopic();
 
+    /**
+     * Get the source config.
+     *
+     * @return source config
+     */
+    SourceConfig getSourceConfig();
+
     /**
      * New output message using schema for serializing to the topic.
      *