Set correct ssl factory when using OAUTH

Author: psainics

src/main/java/io/cdap/plugin/http/common/http/OAuthUtil.java

@@ -83,6 +83,13 @@ public static AccessToken getAccessToken(BaseHttpConfig config) throws IOExcepti
         // get accessToken from service account
         return OAuthUtil.getAccessTokenByServiceAccount(config);
       case OAUTH2:
+        if (config instanceof BaseHttpSourceConfig) {
+          try (CloseableHttpClient client = HttpClients.custom()
+              .setSSLSocketFactory(new SSLConnectionSocketFactoryCreator((BaseHttpSourceConfig) config).create())
+              .build()) {
+            return getAccessToken(client, config);
+          }
+        }
         try (CloseableHttpClient client = HttpClients.createDefault()) {
           return getAccessToken(client, config);
         }

src/main/java/io/cdap/plugin/http/source/batch/HttpBatchSourceConfig.java

@@ -21,6 +21,7 @@
 import io.cdap.plugin.http.common.http.AuthType;
 import io.cdap.plugin.http.common.http.HttpClient;
 import io.cdap.plugin.http.common.http.OAuthUtil;
+import io.cdap.plugin.http.common.http.SSLConnectionSocketFactoryCreator;
 import io.cdap.plugin.http.source.common.BaseHttpSourceConfig;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpHost;
@@ -70,7 +71,8 @@ private void validateOAuth2Credentials(FailureCollector collector) {
       !containsMacro(PROPERTY_PROXY_PASSWORD) && !containsMacro(PROPERTY_PROXY_USERNAME) &&
       !containsMacro(PROPERTY_PROXY_URL) && !containsMacro(PROPERTY_OAUTH2_CLIENT_AUTHENTICATION) &&
       !containsMacro(PROPERTY_OAUTH2_GRANT_TYPE)) {
-      HttpClientBuilder httpclientBuilder = HttpClients.custom();
+      HttpClientBuilder httpclientBuilder = HttpClients.custom()
+          .setSSLSocketFactory(new SSLConnectionSocketFactoryCreator(this).create());
       if (!Strings.isNullOrEmpty(getProxyUrl())) {
         HttpHost proxyHost = HttpHost.create(getProxyUrl());
         if (!Strings.isNullOrEmpty(getProxyUsername()) && !Strings.isNullOrEmpty(getProxyPassword())) {