allow using file:// to manage packages

pkg/connection/reconcile_package.go

@@ -126,12 +126,30 @@ func (r *PulsarPackageReconciler) ReconcilePackage(ctx context.Context, pulsarAd
 		return nil
 	}
 
-	filePath, err := createTmpFile(pkg.Spec.FileURL)
+	parsedFileURL, err := url.Parse(pkg.Spec.FileURL)
 	if err != nil {
-		log.Error(err, "Failed to download the package file")
-		return err
+		return fmt.Errorf("invalid FileURL: %v", err)
+	}
+	if !isSupportedScheme(parsedFileURL.Scheme) {
+		return fmt.Errorf("unsupported scheme: %s", parsedFileURL.Scheme)
+	}
+
+	filePath := ""
+	switch parsedFileURL.Scheme {
+	case "http", "https":
+		filePath, err = createTmpFile(pkg.Spec.FileURL)
+		if err != nil {
+			log.Error(err, "Failed to download the package file")
+			return err
+		}
+		defer os.Remove(filePath)
+	case "file":
+		// check parsedFileUrl.Path is a valid file path
+		if _, err := os.Stat(parsedFileURL.Path); err != nil {
+			return fmt.Errorf("invalid file path: %s", parsedFileURL.Path)
+		}
+		filePath = parsedFileURL.Path
 	}
-	defer os.Remove(filePath)
 
 	updated := false
 	if exist, err := pulsarAdmin.CheckPulsarPackageExist(pkg.Spec.PackageURL); err != nil {
@@ -198,3 +216,7 @@ func createTmpFile(fileURL string) (string, error) {
 
 	return tmpFile.Name(), nil
 }
+
+func isSupportedScheme(scheme string) bool {
+	return scheme == "http" || scheme == "https" || scheme == "file"
+}

tests/operator/resources_test.go

@@ -89,6 +89,7 @@ var _ = Describe("Resources", func() {
 		psource                *v1alphav1.PulsarSource
 		pnsisolationpolicy     *v1alphav1.PulsarNSIsolationPolicy
 		psourcepackageurl      string = "builtin://data-generator"
+		invalidFileURL         string = "file://invalid-file-path"
 	)
 
 	BeforeEach(func() {
@@ -496,6 +497,24 @@ var _ = Describe("Resources", func() {
 			})
 		})
 
+		Context("PulsarPackage operation with invalid file URL", func() {
+			It("should create the pulsarpackage failed with invalid file URL", func() {
+				ppackage.Spec.FileURL = invalidFileURL
+				ppackage.Spec.PackageURL = "function://public/default/file@invalid"
+				err := k8sClient.Create(ctx, ppackage)
+				Expect(err != nil).Should(BeTrue())
+			})
+		})
+
+		Context("PulsarPackage operation with valid file URL", func() {
+			It("should create the pulsarpackage successfully with valid file URL", func() {
+				ppackage.Spec.FileURL = "file:///manager" // we use the manager binary as the file URL
+				ppackage.Spec.PackageURL = "function://public/default/file@valid"
+				err := k8sClient.Create(ctx, ppackage)
+				Expect(err == nil || apierrors.IsAlreadyExists(err)).Should(BeTrue())
+			})
+		})
+
 		AfterAll(func() {
 			Eventually(func(g Gomega) {
 				t := &v1alphav1.PulsarTopic{}