[improve] Allow download link with basic auth (#19750)

Co-authored-by: tison <wander4096@gmail.com>

Author: vanhtuan0409

pulsar-client/src/main/java/org/apache/pulsar/client/impl/auth/AuthenticationDataBasic.java

@@ -27,16 +27,20 @@
 
 public class AuthenticationDataBasic implements AuthenticationDataProvider {
     private static final String HTTP_HEADER_NAME = "Authorization";
-    private String httpAuthToken;
-    private String commandAuthToken;
-    private Map<String, String> headers = new HashMap<>();
+    private final String commandAuthToken;
+    private final Map<String, String> headers;
 
     public AuthenticationDataBasic(String userId, String password) {
-        httpAuthToken = "Basic " + Base64.getEncoder().encodeToString((userId + ":" + password).getBytes());
-        commandAuthToken = userId + ":" + password;
-        headers.put(HTTP_HEADER_NAME, httpAuthToken);
-        headers.put(PULSAR_AUTH_METHOD_NAME, AuthenticationBasic.AUTH_METHOD_NAME);
-        this.headers = Collections.unmodifiableMap(this.headers);
+        this(userId + ":" + password);
+    }
+
+    public AuthenticationDataBasic(String userInfo) {
+        String httpAuthToken = "Basic " + Base64.getEncoder().encodeToString(userInfo.getBytes());
+        this.commandAuthToken = userInfo;
+        this.headers = Collections.unmodifiableMap(new HashMap<String, String>(){{
+            put(HTTP_HEADER_NAME, httpAuthToken);
+            put(PULSAR_AUTH_METHOD_NAME, AuthenticationBasic.AUTH_METHOD_NAME);
+        }});
     }
 
     @Override

pulsar-functions/utils/src/main/java/org/apache/pulsar/functions/utils/FunctionCommon.java

@@ -35,9 +35,11 @@
 import java.net.ServerSocket;
 import java.net.URISyntaxException;
 import java.net.URL;
+import java.net.URLConnection;
 import java.nio.file.Files;
 import java.nio.file.StandardCopyOption;
 import java.util.Collection;
+import java.util.Map;
 import java.util.UUID;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
@@ -48,6 +50,7 @@
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.client.impl.auth.AuthenticationDataBasic;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.apache.pulsar.common.functions.Utils;
 import org.apache.pulsar.common.nar.NarClassLoader;
@@ -245,8 +248,15 @@ public static Class<?> getSinkType(Class sinkClass) {
     }
 
     public static void downloadFromHttpUrl(String destPkgUrl, File targetFile) throws IOException {
-        URL website = new URL(destPkgUrl);
-        try (InputStream in = website.openStream()) {
+        final URL url = new URL(destPkgUrl);
+        final URLConnection connection = url.openConnection();
+        if (StringUtils.isNotEmpty(url.getUserInfo())) {
+            final AuthenticationDataBasic authBasic = new AuthenticationDataBasic(url.getUserInfo());
+            for (Map.Entry<String, String> header : authBasic.getHttpHeaders()) {
+                connection.setRequestProperty(header.getKey(), header.getValue());
+            }
+        }
+        try (InputStream in = connection.getInputStream()) {
             log.info("Downloading function package from {} to {} ...", destPkgUrl, targetFile.getAbsoluteFile());
             Files.copy(in, targetFile.toPath(), StandardCopyOption.REPLACE_EXISTING);
         }

pulsar-functions/utils/src/test/java/org/apache/pulsar/functions/utils/FunctionCommonTest.java

@@ -18,6 +18,11 @@
  */
 package org.apache.pulsar.functions.utils;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import java.io.File;
 import java.util.Collection;
 import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.common.util.FutureUtil;
@@ -26,17 +31,11 @@
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.functions.api.WindowContext;
 import org.apache.pulsar.functions.api.WindowFunction;
+import org.assertj.core.util.Files;
 import org.testng.Assert;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
-import java.io.File;
-import java.util.UUID;
-
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import static org.testng.Assert.assertEquals;
-
 /**
  * Unit test of {@link Exceptions}.
  */
@@ -78,12 +77,22 @@ public void testValidateHttpFileUrl() throws Exception {
 
     @Test
     public void testDownloadFile() throws Exception {
-        String jarHttpUrl = "https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-common/2.4.2/pulsar-common-2.4.2.jar";
-        String testDir = FunctionCommonTest.class.getProtectionDomain().getCodeSource().getLocation().getPath();
-        File pkgFile = new File(testDir, UUID.randomUUID().toString());
-        FunctionCommon.downloadFromHttpUrl(jarHttpUrl, pkgFile);
-        Assert.assertTrue(pkgFile.exists());
-        pkgFile.delete();
+        final String jarHttpUrl = "https://repo1.maven.org/maven2/org/apache/pulsar/pulsar-common/2.4.2/pulsar-common-2.4.2.jar";
+        final File file = Files.newTemporaryFile();
+        file.deleteOnExit();
+        assertThat(file.length()).isZero();
+        FunctionCommon.downloadFromHttpUrl(jarHttpUrl, file);
+        assertThat(file.length()).isGreaterThan(0);
+    }
+
+    @Test
+    public void testDownloadFileWithBasicAuth() throws Exception {
+        final String jarHttpUrl = "https://foo:bar@httpbin.org/basic-auth/foo/bar";
+        final File file = Files.newTemporaryFile();
+        file.deleteOnExit();
+        assertThat(file.length()).isZero();
+        FunctionCommon.downloadFromHttpUrl(jarHttpUrl, file);
+        assertThat(file.length()).isGreaterThan(0);
     }
 
     @Test