diff --git a/.gitignore b/.gitignore
index f6b69f4d8..50e2532d8 100644
--- a/.gitignore
+++ b/.gitignore
@@ -12,4 +12,5 @@ target/
 bin
 .settings
 .gradletasknamecache
-.DS_Store
\ No newline at end of file
+.DS_Store
+dagger-common/src/generated-sources/
\ No newline at end of file
diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnector.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnector.java
index 5e6b786a3..591f1c927 100644
--- a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnector.java
+++ b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnector.java
@@ -98,11 +98,13 @@ protected void process(Row input, ResultFuture<Row> resultFuture) {
                     .getVariablesValue(rowManager, ExternalPostProcessorVariableType.REQUEST_VARIABLES, httpSourceConfig.getRequestVariables(), resultFuture);
             Object[] dynamicHeaderVariablesValues = getEndpointHandler()
                     .getVariablesValue(rowManager, ExternalPostProcessorVariableType.HEADER_VARIABLES, httpSourceConfig.getHeaderVariables(), resultFuture);
+            Object[] endpointVariablesValues = getEndpointHandler()
+                    .getVariablesValue(rowManager, ExternalPostProcessorVariableType.ENDPOINT_VARIABLE, httpSourceConfig.getEndpointVariables(), resultFuture);
             if (getEndpointHandler().isQueryInvalid(resultFuture, rowManager, httpSourceConfig.getRequestVariables(), requestVariablesValues) || getEndpointHandler().isQueryInvalid(resultFuture, rowManager, httpSourceConfig.getHeaderVariables(), dynamicHeaderVariablesValues)) {
                 return;
             }
 
-            BoundRequestBuilder request = HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues, dynamicHeaderVariablesValues);
+            BoundRequestBuilder request = HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues, dynamicHeaderVariablesValues, endpointVariablesValues);
             HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, getMeterStatsManager(),
                     rowManager, getColumnNameManager(), getOutputDescriptor(resultFuture), resultFuture, getErrorReporter(), new PostResponseTelemetry());
             httpResponseHandler.startTimer();
diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfig.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfig.java
index d30efc2ad..9d66cbdb4 100644
--- a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfig.java
+++ b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfig.java
@@ -17,6 +17,7 @@
  */
 public class HttpSourceConfig implements Serializable, SourceConfig {
     private String endpoint;
+    private String endpointVariables;
     private String verb;
     private String requestPattern;
     private String requestVariables;
@@ -39,6 +40,7 @@ public class HttpSourceConfig implements Serializable, SourceConfig {
      * Instantiates a new Http source config.
      *
      * @param endpoint           the endpoint
+     * @param endpointVariables  the endpoint variables
      * @param verb               the verb
      * @param requestPattern     the request pattern
      * @param requestVariables   the request variables
@@ -54,8 +56,9 @@ public class HttpSourceConfig implements Serializable, SourceConfig {
      * @param metricId           the metric id
      * @param retainResponseType the retain response type
      */
-    public HttpSourceConfig(String endpoint, String verb, String requestPattern, String requestVariables, String headerPattern, String headerVariables, String streamTimeout, String connectTimeout, boolean failOnErrors, String type, String capacity, Map<String, String> headers, Map<String, OutputMapping> outputMapping, String metricId, boolean retainResponseType) {
+    public HttpSourceConfig(String endpoint, String endpointVariables, String verb, String requestPattern, String requestVariables, String headerPattern, String headerVariables, String streamTimeout, String connectTimeout, boolean failOnErrors, String type, String capacity, Map<String, String> headers, Map<String, OutputMapping> outputMapping, String metricId, boolean retainResponseType) {
         this.endpoint = endpoint;
+        this.endpointVariables = endpointVariables;
         this.verb = verb;
         this.requestPattern = requestPattern;
         this.requestVariables = requestVariables;
@@ -90,6 +93,16 @@ public String getEndpoint() {
         return endpoint;
     }
 
+    /**
+     * Gets endpoint variables.
+     *
+     * @return the endpointVariables
+     */
+    public String getEndpointVariables() {
+        return endpointVariables;
+    }
+
+
     /**
      * Gets verb.
      *
@@ -237,6 +250,6 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(endpoint, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headers, outputMapping, metricId, retainResponseType);
+        return Objects.hash(endpoint, endpointVariables, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headers, outputMapping, metricId, retainResponseType);
     }
 }
diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandler.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandler.java
index cd7a697fe..da9d8889b 100644
--- a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandler.java
+++ b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandler.java
@@ -19,25 +19,29 @@ public class HttpGetRequestHandler implements HttpRequestHandler {
     private AsyncHttpClient httpClient;
     private Object[] requestVariablesValues;
     private Object[] dynamicHeaderVariablesValues;
+    private Object[] endpointVariablesValues;
 
     /**
      * Instantiates a new Http get request handler.
      *
-     * @param httpSourceConfig       the http source config
-     * @param httpClient             the http client
-     * @param requestVariablesValues the request variables values
+     * @param httpSourceConfig        the http source config
+     * @param httpClient              the http client
+     * @param requestVariablesValues  the request variables values
+     * @param endpointVariablesValues the request variables values
      */
-    public HttpGetRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues) {
+    public HttpGetRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues, Object[] endpointVariablesValues) {
         this.httpSourceConfig = httpSourceConfig;
         this.httpClient = httpClient;
         this.requestVariablesValues = requestVariablesValues;
         this.dynamicHeaderVariablesValues = dynamicHeaderVariablesValues;
+        this.endpointVariablesValues = endpointVariablesValues;
     }
 
     @Override
     public BoundRequestBuilder create() {
         String endpointPath = String.format(httpSourceConfig.getPattern(), requestVariablesValues);
-        String endpoint = httpSourceConfig.getEndpoint();
+        String endpoint = String.format(httpSourceConfig.getEndpoint(), endpointVariablesValues);
+
         String requestEndpoint = endpoint + endpointPath;
         BoundRequestBuilder getRequest = httpClient.prepareGet(requestEndpoint);
         Map<String, String> headers = httpSourceConfig.getHeaders();
diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandler.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandler.java
index 5ca1c1ad3..357373415 100644
--- a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandler.java
+++ b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandler.java
@@ -19,24 +19,28 @@ public class HttpPostRequestHandler implements HttpRequestHandler {
     private AsyncHttpClient httpClient;
     private Object[] requestVariablesValues;
     private Object[] dynamicHeaderVariablesValues;
+    private Object[] endpointVariablesValues;
     /**
      * Instantiates a new Http post request handler.
      *
-     * @param httpSourceConfig       the http source config
-     * @param httpClient             the http client
-     * @param requestVariablesValues the request variables values
+     * @param httpSourceConfig        the http source config
+     * @param httpClient              the http client
+     * @param requestVariablesValues  the request variables values
+     * @param endpointVariablesValues the endpoint variables values
      */
-    public HttpPostRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues) {
+    public HttpPostRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues, Object[] endpointVariablesValues) {
         this.httpSourceConfig = httpSourceConfig;
         this.httpClient = httpClient;
         this.requestVariablesValues = requestVariablesValues;
         this.dynamicHeaderVariablesValues = dynamicHeaderVariablesValues;
+        this.endpointVariablesValues = endpointVariablesValues;
     }
 
     @Override
     public BoundRequestBuilder create() {
         String requestBody = String.format(httpSourceConfig.getPattern(), requestVariablesValues);
-        String endpoint = httpSourceConfig.getEndpoint();
+        String endpoint = String.format(httpSourceConfig.getEndpoint(), endpointVariablesValues);
+
         BoundRequestBuilder postRequest = httpClient
                 .preparePost(endpoint)
                 .setBody(requestBody);
diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPutRequestHandler.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPutRequestHandler.java
new file mode 100644
index 000000000..a7603181c
--- /dev/null
+++ b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpPutRequestHandler.java
@@ -0,0 +1,65 @@
+package io.odpf.dagger.core.processors.external.http.request;
+
+import com.google.gson.Gson;
+import io.netty.util.internal.StringUtil;
+import io.odpf.dagger.core.exception.InvalidConfigurationException;
+import io.odpf.dagger.core.processors.external.http.HttpSourceConfig;
+import org.asynchttpclient.AsyncHttpClient;
+import org.asynchttpclient.BoundRequestBuilder;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UnknownFormatConversionException;
+
+/**
+ * The Http post request handler.
+ */
+public class HttpPutRequestHandler implements HttpRequestHandler {
+    private HttpSourceConfig httpSourceConfig;
+    private AsyncHttpClient httpClient;
+    private Object[] requestVariablesValues;
+    private Object[] dynamicHeaderVariablesValues;
+    private Object[] endpointVariablesValues;
+    /**
+     * Instantiates a new Http post request handler.
+     *
+     * @param httpSourceConfig        the http source config
+     * @param httpClient              the http client
+     * @param requestVariablesValues  the request variables values
+     * @param endpointVariablesValues the endpoint variables values
+     */
+    public HttpPutRequestHandler(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues, Object[] dynamicHeaderVariablesValues, Object[] endpointVariablesValues) {
+        this.httpSourceConfig = httpSourceConfig;
+        this.httpClient = httpClient;
+        this.requestVariablesValues = requestVariablesValues;
+        this.dynamicHeaderVariablesValues = dynamicHeaderVariablesValues;
+        this.endpointVariablesValues = endpointVariablesValues;
+    }
+
+    @Override
+    public BoundRequestBuilder create() {
+        String requestBody = String.format(httpSourceConfig.getPattern(), requestVariablesValues);
+        String endpoint = String.format(httpSourceConfig.getEndpoint(), endpointVariablesValues);
+
+        BoundRequestBuilder putRequest = httpClient
+                .preparePut(endpoint)
+                .setBody(requestBody);
+        Map<String, String> headers = httpSourceConfig.getHeaders();
+        if (!StringUtil.isNullOrEmpty(httpSourceConfig.getHeaderPattern())) {
+            try {
+                String dynamicHeader = String.format(httpSourceConfig.getHeaderPattern(), dynamicHeaderVariablesValues);
+                headers.putAll(new Gson().fromJson(dynamicHeader, HashMap.class));
+            } catch (UnknownFormatConversionException e) {
+                throw new InvalidConfigurationException(String.format("pattern config '%s' is invalid", httpSourceConfig.getHeaderPattern()));
+            } catch (IllegalArgumentException e) {
+                throw new InvalidConfigurationException(String.format("pattern config '%s' is incompatible with the variable config '%s'", httpSourceConfig.getHeaderPattern(), httpSourceConfig.getHeaderVariables()));
+            }
+        }
+        return addHeaders(putRequest, headers);
+    }
+
+    @Override
+    public boolean canCreate() {
+        return httpSourceConfig.getVerb().equalsIgnoreCase("put");
+    }
+}
diff --git a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactory.java b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactory.java
index 1e61129ea..6f713cbef 100644
--- a/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactory.java
+++ b/dagger-core/src/main/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactory.java
@@ -19,11 +19,12 @@ public class HttpRequestFactory {
      * @param requestVariablesValues the request variables values
      * @return the bound request builder
      */
-    public static BoundRequestBuilder createRequest(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues,  Object[] headerVariablesValues) {
+    public static BoundRequestBuilder createRequest(HttpSourceConfig httpSourceConfig, AsyncHttpClient httpClient, Object[] requestVariablesValues,  Object[] headerVariablesValues, Object[] endpointVariablesValues) {
 
         ArrayList<HttpRequestHandler> httpRequestHandlers = new ArrayList<>();
-        httpRequestHandlers.add(new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues));
-        httpRequestHandlers.add(new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues));
+        httpRequestHandlers.add(new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues, endpointVariablesValues));
+        httpRequestHandlers.add(new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues, endpointVariablesValues));
+        httpRequestHandlers.add(new HttpPutRequestHandler(httpSourceConfig, httpClient, requestVariablesValues, headerVariablesValues, endpointVariablesValues));
 
         HttpRequestHandler httpRequestHandler = httpRequestHandlers
                 .stream()
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/PostProcessorConfigTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/PostProcessorConfigTest.java
index 8c5ab8fa8..d09909857 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/PostProcessorConfigTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/PostProcessorConfigTest.java
@@ -64,7 +64,7 @@ public void shouldReturnHttpExternalSourceConfig() {
         outputMapping = new OutputMapping("$.data.tensor.values[0]");
         outputMappings.put("surge_factor", outputMapping);
 
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8000", "post", null, null, null, null, "5000", "5000", true, null, null, headerMap, outputMappings, null, false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8000", "", "post", null, null, null, null, "5000", "5000", true, null, null, headerMap, outputMappings, null, false);
 
         assertEquals(httpSourceConfig, defaultPostProcessorConfig.getExternalSource().getHttpConfig().get(0));
     }
@@ -120,7 +120,7 @@ public void shouldBeEmptyWhenNoneOfTheConfigsExist() {
     @Test
     public void shouldNotBeEmptyWhenExternalSourceHasHttpConfigExist() {
         ArrayList<HttpSourceConfig> http = new ArrayList<>();
-        http.add(new HttpSourceConfig("", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), "metricId_01", false));
+        http.add(new HttpSourceConfig("", "", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), "metricId_01", false));
         ArrayList<EsSourceConfig> es = new ArrayList<>();
         ArrayList<PgSourceConfig> pg = new ArrayList<>();
         ExternalSourceConfig externalSourceConfig = new ExternalSourceConfig(http, es, pg, new ArrayList<>());
@@ -290,4 +290,18 @@ public void shouldParseInternalProcessorConfigForInternalSourceConfig() {
         PostProcessorConfig postProcessorConfig = PostProcessorConfig.parse(configuration);
         assertNotNull(postProcessorConfig.getInternalSource().get(0).getInternalProcessorConfig());
     }
+
+    @Test
+    public void shouldParseEndpointVariablesConfig() {
+        String configuration = "{\"external_source\":{\"es\":[{\"host\":\"localhost\",\"port\":\"9200\",\"output_mapping\":{\"customer_profile\":{\"path\":\"$._source\"}},\"endpoint_pattern\":\"/customers/customer/%s\",\"endpoint_variables\":\"customer_id\",\"retry_timeout\":\"5000\",\"socket_timeout\":\"6000\",\"stream_timeout\":\"5000\",\"type\":\"TestLogMessage\"}],\"http\":[{\"body_column_from_sql\":\"request_body\",\"connect_timeout\":\"5000\",\"endpoint\":\"http://localhost:8000/%s\",\"endpoint_variables\":\"some-id\",\"fail_on_errors\":\"true\",\"headers\":{\"content-type\":\"application/json\"},\"output_mapping\":{\"surge_factor\":{\"path\":\"$.data.tensor.values[0]\"}},\"stream_timeout\":\"5000\",\"verb\":\"put\"}]},\"internal_source\":[{\"output_field\":\"event_timestamp\",\"value\":\"CURRENT_TIMESTAMP\",\"type\":\"function\"},{\"output_field\":\"s2_id_level\",\"value\":\"7\",\"type\":\"constant\"}],\"transformers\":[{\"transformation_arguments\":{\"keyColumnName\":\"s2id\",\"valueColumnName\":\"features\"},\"transformation_class\":\"test.postprocessor.FeatureTransformer\"}]}";
+        PostProcessorConfig postProcessorConfig = PostProcessorConfig.parse(configuration);
+        assertEquals("some-id", postProcessorConfig.getExternalSource().getHttpConfig().get(0).getEndpointVariables());
+    }
+
+    @Test
+    public void shouldParseEmptyEndpointVariablesConfig() {
+        String configuration = "{\"external_source\":{\"es\":[{\"host\":\"localhost\",\"port\":\"9200\",\"output_mapping\":{\"customer_profile\":{\"path\":\"$._source\"}},\"endpoint_pattern\":\"/customers/customer/%s\",\"endpoint_variables\":\"customer_id\",\"retry_timeout\":\"5000\",\"socket_timeout\":\"6000\",\"stream_timeout\":\"5000\",\"type\":\"TestLogMessage\"}],\"http\":[{\"body_column_from_sql\":\"request_body\",\"connect_timeout\":\"5000\",\"endpoint\":\"http://localhost:8000/%s\",\"fail_on_errors\":\"true\",\"headers\":{\"content-type\":\"application/json\"},\"output_mapping\":{\"surge_factor\":{\"path\":\"$.data.tensor.values[0]\"}},\"stream_timeout\":\"5000\",\"verb\":\"put\"}]},\"internal_source\":[{\"output_field\":\"event_timestamp\",\"value\":\"CURRENT_TIMESTAMP\",\"type\":\"function\"},{\"output_field\":\"s2_id_level\",\"value\":\"7\",\"type\":\"constant\"}],\"transformers\":[{\"transformation_arguments\":{\"keyColumnName\":\"s2id\",\"valueColumnName\":\"features\"},\"transformation_class\":\"test.postprocessor.FeatureTransformer\"}]}";
+        PostProcessorConfig postProcessorConfig = PostProcessorConfig.parse(configuration);
+        assertEquals(null, postProcessorConfig.getExternalSource().getHttpConfig().get(0).getEndpointVariables());
+    }
 }
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalPostProcessorTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalPostProcessorTest.java
index f31f3c568..c6d661e7d 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalPostProcessorTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalPostProcessorTest.java
@@ -72,7 +72,7 @@ public void setup() {
         HashMap<String, OutputMapping> httpColumnNames = new HashMap<>();
         httpColumnNames.put("http_field_1", new OutputMapping(""));
         httpColumnNames.put("http_field_2", new OutputMapping(""));
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpColumnNames, "metricId_01", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpColumnNames, "metricId_01", false);
         HashMap<String, OutputMapping> esOutputMapping = new HashMap<>();
         esOutputMapping.put("es_field_1", new OutputMapping(""));
         EsSourceConfig esSourceConfig = new EsSourceConfig("host", "port", "", "", "endpointPattern",
@@ -135,7 +135,7 @@ public void shouldProcessWithRightConfiguration() {
         outputMapping.put("order_id", new OutputMapping("path"));
 
         List<HttpSourceConfig> httpSourceConfigs = new ArrayList<>();
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), outputMapping, "metricId_01", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), outputMapping, "metricId_01", false);
         httpSourceConfigs.add(httpSourceConfig);
 
         List<EsSourceConfig> esSourceConfigs = new ArrayList<>();
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalSourceConfigTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalSourceConfigTest.java
index 177859b60..4a3f7245e 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalSourceConfigTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/ExternalSourceConfigTest.java
@@ -29,7 +29,7 @@ public void setUp() {
         HashMap<String, OutputMapping> httpOutputMapping = new HashMap<>();
         httpOutputMapping.put("http_field_1", new OutputMapping(""));
         httpOutputMapping.put("http_field_2", new OutputMapping(""));
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpOutputMapping, "metricId_01", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("endpoint", "", "POST", "/some/patttern/%s", "variable", "", "", "123", "234", false, "type", "20", new HashMap<>(), httpOutputMapping, "metricId_01", false);
         http = new ArrayList<>();
         http.add(httpSourceConfig);
         es = new ArrayList<>();
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/grpc/GrpcSourceConfigTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/grpc/GrpcSourceConfigTest.java
index 8b7331ecf..d1973c08a 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/grpc/GrpcSourceConfigTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/grpc/GrpcSourceConfigTest.java
@@ -119,13 +119,13 @@ public void hasTypeShouldBeTrueWhenTypeIsPresent() {
 
     @Test
     public void hasTypeShouldBeFalseWhenTypeIsNull() {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", null, "", false, null, "", new HashMap<>(), new HashMap<>(), metricId, false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", "", null, "", false, null, "", new HashMap<>(), new HashMap<>(), metricId, false);
         assertFalse(httpSourceConfig.hasType());
     }
 
     @Test
     public void hasTypeShouldBeFalseWhenTypeIsEmpty() {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), metricId, false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "",  "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), metricId, false);
         assertFalse(httpSourceConfig.hasType());
     }
 
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnectorTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnectorTest.java
index 0d049484c..12b6138a4 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnectorTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpAsyncConnectorTest.java
@@ -96,7 +96,7 @@ public void setUp() {
         when(schemaConfig.getOutputProtoClassName()).thenReturn("OutputProtoMessage");
         externalMetricConfig = new ExternalMetricConfig("metricId-http-01", shutDownPeriod, telemetryEnabled);
 
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}",
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}",
                 "customer_id", "", "", "123", "234", false, httpConfigType, "345",
                 headers, outputMapping, "metricId_02", false);
     }
@@ -187,7 +187,7 @@ public void shouldCompleteExceptionallyIfOutputDescriptorNotFound() throws Excep
     public void shouldCompleteExceptionallyWhenRequestVariableIsInvalid() throws Exception {
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
         String invalidRequestVariable = "invalid_variable";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", invalidRequestVariable, "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", invalidRequestVariable, "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(defaultHttpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
@@ -210,7 +210,7 @@ public void shouldCompleteExceptionallyWhenRequestVariableIsInvalid() throws Exc
     @Test
     public void shouldCompleteExceptionallyWhenEndpointVariableIsEmptyAndRequiredInPattern() throws Exception {
         String emptyRequestVariable = "";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", emptyRequestVariable, "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", emptyRequestVariable, "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
@@ -227,7 +227,7 @@ public void shouldCompleteExceptionallyWhenEndpointVariableIsEmptyAndRequiredInP
     @Test
     public void shouldEnrichWhenEndpointVariableIsEmptyAndNotRequiredInPattern() throws Exception {
         String emptyRequestVariable = "";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"static\"}", emptyRequestVariable, "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"static\"}", emptyRequestVariable, "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(defaultHttpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
 
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
@@ -247,7 +247,7 @@ public void shouldEnrichWhenEndpointVariableIsEmptyAndNotRequiredInPattern() thr
     @Test
     public void shouldCompleteExceptionallyWhenEndpointPatternIsInvalid() throws Exception {
         String invalidRequestPattern = "{\"key\": \"%\"}";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", invalidRequestPattern, "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", invalidRequestPattern, "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
@@ -262,7 +262,7 @@ public void shouldCompleteExceptionallyWhenEndpointPatternIsInvalid() throws Exc
 
     @Test
     public void shouldGetDescriptorFromOutputProtoIfTypeNotGiven() throws Exception {
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, null, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, null, "345", headers, outputMapping, "metricId_02", false);
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
         when(stencilClientOrchestrator.getStencilClient()).thenReturn(stencilClient);
@@ -277,7 +277,7 @@ public void shouldGetDescriptorFromOutputProtoIfTypeNotGiven() throws Exception
 
     @Test
     public void shouldGetDescriptorFromTypeIfGiven() throws Exception {
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, "TestBookingLogMessage", "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, "TestBookingLogMessage", "345", headers, outputMapping, "metricId_02", false);
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
         when(stencilClientOrchestrator.getStencilClient()).thenReturn(stencilClient);
@@ -293,7 +293,7 @@ public void shouldGetDescriptorFromTypeIfGiven() throws Exception {
     @Test
     public void shouldCompleteExceptionallyWhenEndpointPatternIsIncompatible() throws Exception {
         String invalidRequestPattern = "{\"key\": \"%d\"}";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", invalidRequestPattern, "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", invalidRequestPattern, "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
@@ -343,7 +343,7 @@ public void shouldAddDynamicHeaders() throws Exception {
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
         when(stencilClientOrchestrator.getStencilClient()).thenReturn(stencilClient);
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
-        HttpSourceConfig dynamicHeaderHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}",
+        HttpSourceConfig dynamicHeaderHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}",
                 "customer_id", "{\"X_KEY\": \"%s\"}", "customer_id", "123", "234", false, httpConfigType, "345",
                 headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(dynamicHeaderHttpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
@@ -361,7 +361,7 @@ public void shouldNotAddDynamicHeaders() throws Exception {
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
         when(stencilClientOrchestrator.getStencilClient()).thenReturn(stencilClient);
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
-        HttpSourceConfig dynamicHeaderHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}",
+        HttpSourceConfig dynamicHeaderHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}",
                 "customer_id", "{\"X_KEY\": \"%s\"}", "customer_ids", "123", "234", false, httpConfigType, "345",
                 headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(dynamicHeaderHttpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
@@ -377,7 +377,7 @@ public void shouldNotAddDynamicHeaders() throws Exception {
     public void shouldCompleteExceptionallyWhenHeaderVariableIsInvalid() throws Exception {
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
         String invalidHeaderVariable = "invalid_variable";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "{\"key\": \"%s\"}", invalidHeaderVariable, "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "{\"key\": \"%s\"}", invalidHeaderVariable, "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(defaultHttpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
@@ -400,7 +400,7 @@ public void shouldCompleteExceptionallyWhenHeaderVariableIsInvalid() throws Exce
     @Test
     public void shouldCompleteExceptionallyWhenHeaderPatternIsIncompatible() throws Exception {
         String invalidHeaderPattern = "{\"key\": \"%d\"}";
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", invalidHeaderPattern, "customer_id", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", invalidHeaderPattern, "customer_id", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(boundRequestBuilder);
         when(boundRequestBuilder.setBody("{\"key\": \"123456\"}")).thenReturn(boundRequestBuilder);
@@ -415,7 +415,7 @@ public void shouldCompleteExceptionallyWhenHeaderPatternIsIncompatible() throws
 
     @Test
     public void shouldThrowExceptionInTimeoutIfFailOnErrorIsTrue() throws Exception {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(httpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
 
         httpAsyncConnector.timeout(streamData, resultFuture);
@@ -425,7 +425,7 @@ public void shouldThrowExceptionInTimeoutIfFailOnErrorIsTrue() throws Exception
 
     @Test
     public void shouldReportFatalInTimeoutIfFailOnErrorIsTrue() throws Exception {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(httpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
 
         httpAsyncConnector.timeout(streamData, resultFuture);
@@ -435,7 +435,7 @@ public void shouldReportFatalInTimeoutIfFailOnErrorIsTrue() throws Exception {
 
     @Test
     public void shouldReportNonFatalInTimeoutIfFailOnErrorIsFalse() {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(httpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
 
         httpAsyncConnector.timeout(streamData, resultFuture);
@@ -455,7 +455,7 @@ public void shouldPassTheInputWithRowSizeCorrespondingToColumnNamesInTimeoutIfFa
     public void shouldThrowExceptionIfUnsupportedHttpVerbProvided() throws Exception {
         when(defaultDescriptorManager.getDescriptor(inputProtoClasses[0])).thenReturn(TestBookingLogMessage.getDescriptor());
 
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "PATCH", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "PATCH", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpAsyncConnector httpAsyncConnector = new HttpAsyncConnector(httpSourceConfig, externalMetricConfig, schemaConfig, httpClient, errorReporter, meterStatsManager, defaultDescriptorManager);
 
         httpAsyncConnector.open(flinkConfiguration);
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpResponseHandlerTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpResponseHandlerTest.java
index 16db27da2..aecfabf83 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpResponseHandlerTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpResponseHandlerTest.java
@@ -74,7 +74,7 @@ public void setup() {
         streamData.setField(1, new Row(2));
         rowManager = new RowManager(streamData);
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
     }
 
     @Test
@@ -148,7 +148,7 @@ public void shouldPassInputIfFailOnErrorFalseAndStatusCodeIsOtherThan5XXAnd4XX()
 
     @Test
     public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIs404() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         when(response.getStatusCode()).thenReturn(404);
 
@@ -167,7 +167,7 @@ public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIs404() {
 
     @Test
     public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIs4XXOtherThan404() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         when(response.getStatusCode()).thenReturn(400);
 
@@ -185,7 +185,7 @@ public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIs4XXOtherThan404() {
 
     @Test
     public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIs5XX() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         when(response.getStatusCode()).thenReturn(502);
 
@@ -203,7 +203,7 @@ public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIs5XX() {
 
     @Test
     public void shouldThrowErrorIfFailOnErrorTrueAndStatusCodeIsOtherThan5XXAnd4XX() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         when(response.getStatusCode()).thenReturn(302);
 
@@ -237,7 +237,7 @@ public void shouldPassInputIfFailOnErrorFalseAndOnThrowable() {
 
     @Test
     public void shouldThrowErrorIfFailOnErrorTrueAndOnThrowable() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Throwable throwable = new Throwable("throwable message");
 
@@ -258,7 +258,7 @@ public void shouldPopulateSingleResultFromHttpCallInInputRow() {
         outputMapping.put("surge_factor", new OutputMapping("$.surge"));
         outputColumnNames = Collections.singletonList("surge_factor");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Row resultStreamData = new Row(2);
         Row outputData = new Row(2);
@@ -284,7 +284,7 @@ public void shouldPopulateMultipleResultsFromHttpCallInInputRow() {
         outputMapping.put("s2_id_level", new OutputMapping("$.prediction"));
         outputColumnNames = Arrays.asList("surge_factor", "s2_id_level");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Row resultStreamData = new Row(2);
         Row outputData = new Row(2);
@@ -313,7 +313,7 @@ public void shouldThrowExceptionIfFieldNotFoundInFieldDescriptorWhenTypeIsPassed
         outputMapping.put("surge_factor", new OutputMapping("$.surge"));
         outputColumnNames = Collections.singletonList("surge_factor");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         when(response.getStatusCode()).thenReturn(200);
         when(response.getResponseBody()).thenReturn("{\n"
                 + "  \"surge\": 0.732\n"
@@ -331,7 +331,7 @@ public void shouldThrowExceptionIfPathIsWrongIfFailOnErrorsTrue() {
         outputMapping.put("surge_factor", new OutputMapping("invalidPath"));
         outputColumnNames = Collections.singletonList("surge_factor");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", true, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Row resultStreamData = new Row(2);
         Row outputData = new Row(2);
@@ -356,7 +356,7 @@ public void shouldPopulateResultAsObjectIfTypeIsNotPassedAndRetainResponseTypeIs
         outputMapping.put("surge_factor", new OutputMapping("$.surge"));
         outputColumnNames = Collections.singletonList("surge_factor");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, null, "345", headers, outputMapping, "metricId_02", true);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, null, "345", headers, outputMapping, "metricId_02", true);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Row resultStreamData = new Row(2);
         Row outputData = new Row(2);
@@ -381,7 +381,7 @@ public void shouldNotPopulateResultAsObjectIfTypeIsNotPassedAndRetainResponseTyp
         outputMapping.put("surge_factor", new OutputMapping("$.surge"));
         outputColumnNames = Collections.singletonList("surge_factor");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, null, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, null, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Row resultStreamData = new Row(2);
         Row outputData = new Row(2);
@@ -406,7 +406,7 @@ public void shouldHandleAnySuccessResponseCodeOtherThan200() {
         outputMapping.put("surge_factor", new OutputMapping("$.surge"));
         outputColumnNames = Collections.singletonList("surge_factor");
         columnNameManager = new ColumnNameManager(inputColumnNames, outputColumnNames);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "customer_id", "", "", "123", "234", false, httpConfigType, "345", headers, outputMapping, "metricId_02", false);
         HttpResponseHandler httpResponseHandler = new HttpResponseHandler(httpSourceConfig, meterStatsManager, rowManager, columnNameManager, descriptor, resultFuture, errorReporter, new PostResponseTelemetry());
         Row resultStreamData = new Row(2);
         Row outputData = new Row(2);
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfigTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfigTest.java
index 6ee87cc1f..736fe022d 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfigTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/HttpSourceConfigTest.java
@@ -19,6 +19,7 @@ public class HttpSourceConfigTest {
     private OutputMapping outputMapping;
     private String streamTimeout;
     private String endpoint;
+    private String endpointVariable;
     private String verb;
     private String requestPattern;
     private String requestVariables;
@@ -40,6 +41,7 @@ public void setup() {
         outputMappings.put("surge_factor", outputMapping);
         streamTimeout = "123";
         endpoint = "http://localhost:1234";
+        endpointVariable = "";
         verb = "POST";
         requestPattern = "/customers/customer/%s";
         requestVariables = "customer_id";
@@ -51,7 +53,7 @@ public void setup() {
         capacity = "345";
         metricId = "metricId-http-01";
         retainResponseType = false;
-        defaultHttpSourceConfig = new HttpSourceConfig(endpoint, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headerMap, outputMappings, metricId, retainResponseType);
+        defaultHttpSourceConfig = new HttpSourceConfig(endpoint, endpointVariable, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headerMap, outputMappings, metricId, retainResponseType);
     }
 
     @Test
@@ -111,13 +113,13 @@ public void hasTypeShouldBeTrueWhenTypeIsPresent() {
 
     @Test
     public void hasTypeShouldBeFalseWhenTypeIsNull() {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", null, "", false, null, "", new HashMap<>(), new HashMap<>(), metricId, false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", "", null, "", false, null, "", new HashMap<>(), new HashMap<>(), metricId, false);
         assertFalse(httpSourceConfig.hasType());
     }
 
     @Test
     public void hasTypeShouldBeFalseWhenTypeIsEmpty() {
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), metricId, false);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("", "", "", "", "", "", "", "", "", false, "", "", new HashMap<>(), new HashMap<>(), metricId, false);
         assertFalse(httpSourceConfig.hasType());
     }
 
@@ -151,7 +153,7 @@ public void shouldValidate() {
     @Test
     public void shouldThrowExceptionIfAllFieldsMissing() {
 
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig(null, null, null, requestVariables, null, null, null, null, false, null, capacity, null, null, metricId, retainResponseType);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig(null, null, null, null, requestVariables, null, null, null, null, false, null, capacity, null, null, metricId, retainResponseType);
         IllegalArgumentException exception =
                 assertThrows(IllegalArgumentException.class, () -> httpSourceConfig.validateFields());
         assertEquals("Missing required fields: [endpoint, streamTimeout, requestPattern, verb, connectTimeout, outputMapping]", exception.getMessage());
@@ -160,7 +162,7 @@ public void shouldThrowExceptionIfAllFieldsMissing() {
     @Test
     public void shouldThrowExceptionIfSomeFieldsMissing() {
 
-        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("localhost", "post", "body", requestVariables, null, null, null, null, false, null, capacity, null, null, "metricId_01", retainResponseType);
+        HttpSourceConfig httpSourceConfig = new HttpSourceConfig("localhost", "", "post", "body", requestVariables, null, null, null, null, false, null, capacity, null, null, "metricId_01", retainResponseType);
         IllegalArgumentException exception =
                 assertThrows(IllegalArgumentException.class, () -> httpSourceConfig.validateFields());
         assertEquals("Missing required fields: [streamTimeout, connectTimeout, outputMapping]", exception.getMessage());
@@ -173,7 +175,7 @@ public void shouldThrowExceptionIfFieldsOfNestedObjectsAreMissing() {
 
         outputMappings.put("field", outputMappingWithNullField);
 
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost",
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost", "",
                 "post", "request_body", requestVariables, "", "", "4000", "1000", false, "", capacity, headerMap, outputMappings, "metricId_01", retainResponseType);
         IllegalArgumentException exception =
                 assertThrows(IllegalArgumentException.class, () -> defaultHttpSourceConfig.validateFields());
@@ -188,7 +190,7 @@ public void shouldThrowExceptionIfRequestPatternIsEmpty() {
 
         outputMappings.put("field", outputMappingWithNullField);
 
-        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost",
+        defaultHttpSourceConfig = new HttpSourceConfig("http://localhost", "",
                 "post", "", requestVariables, "", "", "4000", "1000", false, "", capacity, headerMap, outputMappings, "metricId_01", retainResponseType);
         IllegalArgumentException exception =
                 assertThrows(IllegalArgumentException.class, () -> defaultHttpSourceConfig.validateFields());
@@ -224,7 +226,7 @@ public void shouldReturnMandatoryFields() {
     @Test
     public void shouldValidateWhenOutputMappingIsEmpty() {
 
-        defaultHttpSourceConfig = new HttpSourceConfig(endpoint, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headerMap, new HashMap<>(), "metricId_01", retainResponseType);
+        defaultHttpSourceConfig = new HttpSourceConfig(endpoint, endpointVariable, verb, requestPattern, requestVariables, headerPattern, headerVariables, streamTimeout, connectTimeout, failOnErrors, type, capacity, headerMap, new HashMap<>(), "metricId_01", retainResponseType);
 
         IllegalArgumentException exception =
                 assertThrows(IllegalArgumentException.class, () -> defaultHttpSourceConfig.validateFields());
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandlerTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandlerTest.java
index afe6d3579..dc6da7f72 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandlerTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpGetRequestHandlerTest.java
@@ -25,6 +25,7 @@ public class HttpGetRequestHandlerTest {
     private HttpSourceConfig httpSourceConfig;
     private ArrayList<Object> requestVariablesValues;
     private ArrayList<Object> headerVariablesValues;
+    private ArrayList<Object> endpointVariablesValues;
 
     @Before
     public void setup() {
@@ -34,27 +35,28 @@ public void setup() {
         headerVariablesValues = new ArrayList<>();
         headerVariablesValues.add("1");
         headerVariablesValues.add("2");
+        endpointVariablesValues = new ArrayList<>();
     }
 
     @Test
     public void shouldReturnTrueForGetVerbOnCanCreate() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         assertTrue(httpGetRequestBuilder.canCreate());
     }
 
     @Test
     public void shouldReturnFalseForVerbOtherThanGetOnCanBuild() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         assertFalse(httpGetRequestBuilder.canCreate());
     }
 
     @Test
     public void shouldBuildGetRequest() {
         when(httpClient.prepareGet("http://localhost:8080/test/key/1")).thenReturn(request);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         assertEquals(request, httpGetRequestBuilder.create());
     }
 
@@ -62,8 +64,8 @@ public void shouldBuildGetRequest() {
     public void shouldBuildGetRequestWithOnlyDynamicHeader() {
         when(httpClient.prepareGet("http://localhost:8080/test/key/1")).thenReturn(request);
         when(request.addHeader("header_key", "1")).thenReturn(request);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "/key/%s", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         httpGetRequestBuilder.create();
         verify(request, times(1)).addHeader(anyString(), anyString());
         verify(request, times(1)).addHeader("header_key", "1");
@@ -75,8 +77,8 @@ public void shouldBuildGetRequestWithDynamicAndStaticHeader() {
         when(request.addHeader("header_key", "1")).thenReturn(request);
         HashMap<String, String> staticHeader = new HashMap<String, String>();
         staticHeader.put("static", "2");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "/key/%s", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         httpGetRequestBuilder.create();
         verify(request, times(2)).addHeader(anyString(), anyString());
         verify(request, times(1)).addHeader("header_key", "1");
@@ -88,8 +90,8 @@ public void shouldBuildGetRequestWithMultipleDynamicAndStaticHeaders() {
         when(httpClient.prepareGet("http://localhost:8080/test/key/1")).thenReturn(request);
         HashMap<String, String> staticHeader = new HashMap<String, String>();
         staticHeader.put("static", "3");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%s\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "/key/%s", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%s\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         httpGetRequestBuilder.create();
         verify(request, times(3)).addHeader(anyString(), anyString());
         verify(request, times(1)).addHeader("header_key_1", "1");
@@ -105,8 +107,8 @@ public void shouldThrowErrorIfHeaderVariablesAreIncompatible() {
         ArrayList incompatibleHeaderVariablesValues = new ArrayList<>();
         incompatibleHeaderVariablesValues.add("test1");
         incompatibleHeaderVariablesValues.add("test12");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%d\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), incompatibleHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "/key/%s", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%d\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), incompatibleHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         InvalidConfigurationException exception = assertThrows(InvalidConfigurationException.class, () -> httpGetRequestBuilder.create());
         assertEquals("pattern config '{\"header_key_1\": \"%s\",\"header_key_2\": \"%d\"}' is incompatible with the variable config '1,2'", exception.getMessage());
     }
@@ -116,8 +118,8 @@ public void shouldThrowErrorIfHeaderHeaderPatternIsInvalid() {
         when(httpClient.prepareGet("http://localhost:8080/test/key/1")).thenReturn(request);
         HashMap<String, String> staticHeader = new HashMap<String, String>();
         staticHeader.put("static", "3");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%p\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "/key/%s", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%p\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpGetRequestHandler httpGetRequestBuilder = new HttpGetRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
         InvalidConfigurationException exception = assertThrows(InvalidConfigurationException.class, () -> httpGetRequestBuilder.create());
         assertEquals("pattern config '{\"header_key_1\": \"%s\",\"header_key_2\": \"%p\"}' is invalid", exception.getMessage());
     }
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandlerTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandlerTest.java
index a5c2f89ca..95faf4d6c 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandlerTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpPostRequestHandlerTest.java
@@ -27,6 +27,7 @@ public class HttpPostRequestHandlerTest {
     private ArrayList<Object> requestVariablesValues;
     private ArrayList<Object> dynamicHeaderVariablesValues;
 
+    private ArrayList<Object> endpointVariablesValues;
     @Before
     public void setup() {
         initMocks(this);
@@ -35,19 +36,20 @@ public void setup() {
         dynamicHeaderVariablesValues = new ArrayList<>();
         dynamicHeaderVariablesValues.add("1");
         dynamicHeaderVariablesValues.add("2");
+        endpointVariablesValues = new ArrayList<>();
     }
 
     @Test
     public void shouldReturnTrueForPostVerbOnCanCreate() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", null, "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         assertTrue(httpPostRequestBuilder.canCreate());
     }
 
     @Test
     public void shouldReturnFalseForVerbOtherThanPostOnCanBuild() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "GET", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         assertFalse(httpPostRequestBuilder.canCreate());
     }
 
@@ -55,8 +57,8 @@ public void shouldReturnFalseForVerbOtherThanPostOnCanBuild() {
     public void shouldBuildPostRequestWithoutHeader() {
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(request);
         when(request.setBody("{\"key\": \"1\"}")).thenReturn(request);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         assertEquals(request, httpPostRequestBuilder.create());
     }
 
@@ -64,8 +66,8 @@ public void shouldBuildPostRequestWithoutHeader() {
     public void shouldBuildPostRequestWithOnlyDynamicHeader() {
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(request);
         when(request.setBody("{\"key\": \"1\"}")).thenReturn(request);
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         httpPostRequestBuilder.create();
         verify(request, times(1)).addHeader(anyString(), anyString());
         verify(request, times(1)).addHeader("header_key", "1");
@@ -77,8 +79,8 @@ public void shouldBuildPostRequestWithDynamicAndStaticHeader() {
         when(request.setBody("{\"key\": \"1\"}")).thenReturn(request);
         HashMap<String, String> staticHeader = new HashMap<String, String>();
         staticHeader.put("static", "2");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key\": \"%s\"}", "1", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         httpPostRequestBuilder.create();
         verify(request, times(2)).addHeader(anyString(), anyString());
         verify(request, times(1)).addHeader("header_key", "1");
@@ -91,8 +93,8 @@ public void shouldBuildPostRequestWithMultipleDynamicAndStaticHeaders() {
         when(request.setBody("{\"key\": \"1\"}")).thenReturn(request);
         HashMap<String, String> staticHeader = new HashMap<String, String>();
         staticHeader.put("static", "3");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%s\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%s\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         httpPostRequestBuilder.create();
         verify(request, times(3)).addHeader(anyString(), anyString());
         verify(request, times(1)).addHeader("header_key_1", "1");
@@ -109,8 +111,8 @@ public void shouldThrowErrorIfHeaderVariablesAreIncompatible() {
         ArrayList incompatibleHeaderVariablesValues = new ArrayList<>();
         incompatibleHeaderVariablesValues.add("test1");
         incompatibleHeaderVariablesValues.add("test12");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%d\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), incompatibleHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%d\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), incompatibleHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         InvalidConfigurationException exception = assertThrows(InvalidConfigurationException.class, () -> httpPostRequestBuilder.create());
         assertEquals("pattern config '{\"header_key_1\": \"%s\",\"header_key_2\": \"%d\"}' is incompatible with the variable config '1,2'", exception.getMessage());
     }
@@ -121,8 +123,8 @@ public void shouldThrowErrorIfHeaderHeaderPatternIsInvalid() {
         when(request.setBody("{\"key\": \"1\"}")).thenReturn(request);
         HashMap<String, String> staticHeader = new HashMap<String, String>();
         staticHeader.put("static", "3");
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%p\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
-        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray());
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "POST", "{\"key\": \"%s\"}", "1", "{\"header_key_1\": \"%s\",\"header_key_2\": \"%p\"}", "1,2", "123", "234", false, "type", "345", staticHeader, null, "metricId_01", false);
+        HttpPostRequestHandler httpPostRequestBuilder = new HttpPostRequestHandler(httpSourceConfig, httpClient, requestVariablesValues.toArray(), dynamicHeaderVariablesValues.toArray(), endpointVariablesValues.toArray());
         InvalidConfigurationException exception = assertThrows(InvalidConfigurationException.class, () -> httpPostRequestBuilder.create());
         assertEquals("pattern config '{\"header_key_1\": \"%s\",\"header_key_2\": \"%p\"}' is invalid", exception.getMessage());
     }
diff --git a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactoryTest.java b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactoryTest.java
index 0a6144954..ecd779072 100644
--- a/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactoryTest.java
+++ b/dagger-core/src/test/java/io/odpf/dagger/core/processors/external/http/request/HttpRequestFactoryTest.java
@@ -26,6 +26,7 @@ public class HttpRequestFactoryTest {
     private HttpSourceConfig httpSourceConfig;
     private ArrayList<Object> requestVariablesValues;
     private ArrayList<Object> headerVariablesValues;
+    private ArrayList<Object> endpointVariablesValues;
     private boolean retainResponseType;
 
     @Before
@@ -34,34 +35,91 @@ public void setup() {
         requestVariablesValues = new ArrayList<>();
         headerVariablesValues = new ArrayList<>();
         requestVariablesValues.add(1);
+        endpointVariablesValues = new ArrayList<>();
         retainResponseType = false;
     }
 
     @Test
     public void shouldReturnPostRequestOnTheBasisOfConfiguration() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", null, "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
         when(httpClient.preparePost("http://localhost:8080/test")).thenReturn(request);
         when(request.setBody("{\"key\": \"123456\"}")).thenReturn(request);
-        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
 
         verify(httpClient, times(1)).preparePost("http://localhost:8080/test");
         verify(httpClient, times(0)).prepareGet(any(String.class));
+        verify(httpClient, times(0)).preparePut(any(String.class));
+    }
+
+    @Test
+    public void shouldReturnPostRequestWithMultiEndpointVariablesOnTheBasisOfConfiguration() {
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test/%s/%s", "exp, 222", "POST", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        when(httpClient.preparePost("http://localhost:8080/test/exp/222")).thenReturn(request);
+        when(request.setBody("{\"key\": \"123456\"}")).thenReturn(request);
+        endpointVariablesValues.add("exp");
+        endpointVariablesValues.add("222");
+        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
+
+        verify(httpClient, times(1)).preparePost("http://localhost:8080/test/exp/222");
+        verify(httpClient, times(0)).prepareGet(any(String.class));
+        verify(httpClient, times(0)).preparePut(any(String.class));
     }
 
     @Test
     public void shouldReturnGetRequestOnTheBasisOfConfiguration() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "GET", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", null, "GET", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
         when(httpClient.prepareGet("http://localhost:8080/test/key/1")).thenReturn(request);
-        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray());
+        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
 
         verify(httpClient, times(1)).prepareGet("http://localhost:8080/test/key/1");
         verify(httpClient, times(0)).preparePost(any(String.class));
+        verify(httpClient, times(0)).preparePut(any(String.class));
+    }
+
+    @Test
+    public void shouldReturnGetRequestWithMultiEndpointVariablesOnTheBasisOfConfiguration() {
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test/%s/%s", "123, 332", "GET", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        when(httpClient.prepareGet("http://localhost:8080/test/key/1")).thenReturn(request);
+        endpointVariablesValues.add("123");
+        endpointVariablesValues.add("332");
+        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
+        verify(httpClient, times(1)).prepareGet("http://localhost:8080/test/123/332/key/1");
+        verify(httpClient, times(0)).preparePost(any(String.class));
+        verify(httpClient, times(0)).preparePut(any(String.class));
+    }
+
+    @Test
+    public void shouldReturnPutRequestOnTheBasisOfConfiguration() {
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test/%s", "123", "PUT", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        when(httpClient.preparePut("http://localhost:8080/test/123")).thenReturn(request);
+        when(request.setBody("{\"key\": \"123456\"}")).thenReturn(request);
+        endpointVariablesValues.add("123");
+        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
+
+        verify(httpClient, times(1)).preparePut("http://localhost:8080/test/123");
+        verify(httpClient, times(0)).prepareGet(any(String.class));
+        verify(httpClient, times(0)).preparePost(any(String.class));
     }
 
+    @Test
+    public void shouldReturnPutRequestWithMultiEndpointVariablesOnTheBasisOfConfiguration() {
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test/%s/abc/%s", "123, 321, asd", "PUT", "{\"key\": \"%s\"}", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        when(httpClient.preparePut("http://localhost:8080/test/123/abc/asd")).thenReturn(request);
+        when(request.setBody("{\"key\": \"123456\"}")).thenReturn(request);
+        endpointVariablesValues.add("123");
+        endpointVariablesValues.add("asd");
+        HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray());
+
+        verify(httpClient, times(1)).preparePut("http://localhost:8080/test/123/abc/asd");
+        verify(httpClient, times(0)).prepareGet(any(String.class));
+        verify(httpClient, times(0)).preparePost(any(String.class));
+    }
+
+
     @Test
     public void shouldThrowExceptionForUnsupportedHttpVerb() {
-        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "PATCH", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
-        assertThrows(InvalidHttpVerbException.class, () -> HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray()));
+        httpSourceConfig = new HttpSourceConfig("http://localhost:8080/test", "", "PATCH", "/key/%s", "1", "", "", "123", "234", false, "type", "345", new HashMap<>(), null, "metricId_01", retainResponseType);
+        assertThrows(InvalidHttpVerbException.class, () -> HttpRequestFactory.createRequest(httpSourceConfig, httpClient, requestVariablesValues.toArray(), headerVariablesValues.toArray(), endpointVariablesValues.toArray()));
     }
 
 }
diff --git a/docs/docs/advance/post_processor.md b/docs/docs/advance/post_processor.md
index 564da0a7a..647dee97f 100644
--- a/docs/docs/advance/post_processor.md
+++ b/docs/docs/advance/post_processor.md
@@ -202,7 +202,7 @@ PROCESSOR_POSTPROCESSOR_CONFIG = {
 
 ### **HTTP**
 
-HTTP Post Processor connects to an external API endpoint and does enrichment based on data from the response of the API call. Currently, we support POST and GET verbs for the API call.
+HTTP Post Processor connects to an external API endpoint and does enrichment based on data from the response of the API call. Currently, we support POST, PUT and GET verbs for the API call.
 
 #### Workflow
 
@@ -210,7 +210,7 @@ On applying only this post processor, dagger will perform the following operatio
 
 - Consume the message from configured Kafka stream.
 - Apply the SQL query configured.
-- Generate the endpoint URL in case of GET call or request body in case of POST call using [request_pattern](post_processor.md#request_pattern) and [request_variables](post_processor.md#request_variables).
+- Generate the endpoint URL in case of GET call or request body in case of POST/PUT call using [request_pattern](post_processor.md#request_pattern) and [request_variables](post_processor.md#request_variables).
 - Make the HTTP call.
 - Read the response from HTTP API and populate the message according to [output_mapping](post_processor.md#output_mapping-1).
 - Push the enriched message to configured sink.
@@ -221,21 +221,29 @@ Following variables need to be configured as part of [PROCESSOR_POSTPROCESSOR_CO
 
 ##### `endpoint`
 
-API endpoint. For POST call, URL will be just the endpoint. For GET, URL will be a combination of endpoint and request pattern.
+API endpoint. For POST and PUT call, URL will be a combination of endpoint and endpoint_variables. For GET, URL will be a combination of endpoint, endpoint_variables and request pattern.
 
-- Example value: `http://127.0.0.1/api/customer`
+- Example value: `http://127.0.0.1/api/customer` or `http://127.0.0.1/api/customer/%s`
 - Type: `required`
 
+##### `endpoint_variables`
+
+List of comma-separated parameters to be replaced in endpoint, these variables must be present in the input proto and selected via the SQL query. Endpoint should also contains %s.
+
+- Example value: `customer_id`
+- Type: `optional`
+
+
 ##### `verb`
 
-HTTP verb (currently support POST and GET).
+HTTP verb (currently support POST, PUT and GET).
 
 - Example value: `GET`
 - Type: `required`
 
 ##### `request_pattern`
 
-Template for the body in case of POST and endpoint path in case of GET.
+Template for the body in case of POST/PUT and endpoint path in case of GET.
 
 - Example value: `/customers/customer/%s`
 - Type: `required`
diff --git a/version.txt b/version.txt
index b1d7abc0d..a0a15177f 100644
--- a/version.txt
+++ b/version.txt
@@ -1 +1 @@
-0.6.2
\ No newline at end of file
+0.6.3
\ No newline at end of file
