feat(warehouse): added base support for s3 datalake deletion as part …

…of regulation API. (#2515)

* Added support for s3 datalake deletion

go.mod

@@ -186,7 +186,7 @@ require (
 	github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f // indirect
 	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
 	github.com/xeipuuv/gojsonschema v1.2.0 // indirect
-	github.com/xitongsys/parquet-go-source v0.0.0-20220803203939-583c0659c569 // indirect
+	github.com/xitongsys/parquet-go-source v0.0.0-20220803203939-583c0659c569
 	github.com/xrash/smetrics v0.0.0-20201216005158-039620a65673 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.5 // indirect
 	go.opencensus.io v0.23.0 // indirect

regulation-worker/cmd/main.go

@@ -69,7 +69,8 @@ func Run(ctx context.Context) {
 		Deleter: delete.NewRouter(
 			&kvstore.KVDeleteManager{},
 			&batch.BatchManager{
-				FMFactory: &filemanager.FileManagerFactoryT{},
+				FMFactory:  &filemanager.FileManagerFactoryT{},
+				FilesLimit: config.GetInt("REGULATION_WORKER_FILES_LIMIT", 1000),
 			},
 			&api.APIManager{
 				Client:           &http.Client{Timeout: config.GetDuration("HttpClient.regulationWorker.timeout", 30, time.Second)},

regulation-worker/internal/client/client.go

@@ -33,12 +33,10 @@ func (j *JobAPI) Get(ctx context.Context) (model.Job, error) {
 	ctx, cancel := context.WithTimeout(ctx, time.Minute)
 	defer cancel()
 
-	method := "GET"
-	genEndPoint := "/dataplane/workspaces/{workspace_id}/regulations/workerJobs"
-	url := fmt.Sprint(j.URLPrefix, prepURL(genEndPoint, j.WorkspaceID))
+	url := fmt.Sprintf("%s/dataplane/workspaces/%s/regulations/workerJobs", j.URLPrefix, j.WorkspaceID)
 	pkgLogger.Debugf("making GET request to URL: %v", url)
 
-	req, err := http.NewRequestWithContext(ctx, method, url, nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		pkgLogger.Errorf("error while create new http request: %v", err)
 		return model.Job{}, err