update legacy producer folder

Author: Ian-Nara

conf/default-config.json

@@ -24,6 +24,8 @@
   "optout_delete_expired": true,
   "optout_s3_bucket": null,
   "optout_s3_folder": null,
+  "optout_sqs_queue_url": null,
+  "optout_sqs_s3_folder": null,
   "cloud_download_threads": 8,
   "cloud_upload_threads": 2,
   "cloud_refresh_interval": 60,
@@ -34,8 +36,7 @@
   "operator_type": "public",
   "uid_instance_id_prefix": "local-optout",
   "optout_enqueue_sqs_enabled": false,
-  "optout_sqs_queue_url": null,
-  "optout_sqs_s3_folder": "sqs-delta",
+  "optout_legacy_producer_s3_folder": null,
   "optout_sqs_max_queue_size": 0,
   "optout_sqs_max_messages_per_poll": 10,
   "optout_sqs_visibility_timeout": 300,

src/main/java/com/uid2/optout/Const.java

@@ -19,7 +19,8 @@ public static class Config extends com.uid2.shared.Const.Config {
         public static final String PartnersMetadataPathProp = "partners_metadata_path";
         public static final String OptOutSqsQueueUrlProp = "optout_sqs_queue_url";
         public static final String OptOutSqsEnabledProp = "optout_enqueue_sqs_enabled";
-        public static final String OptOutSqsS3FolderProp = "optout_sqs_s3_folder"; // Default: "sqs-delta" - folder within same S3 bucket as regular optout
+        public static final String OptOutSqsS3FolderProp = "optout_sqs_s3_folder"; // sqs delta producer writes to this folder
+        public static final String OptOutLegacyProducerS3FolderProp = "optout_legacy_producer_s3_folder"; // legacy producer writes to this folder
         public static final String OptOutSqsMaxMessagesPerPollProp = "optout_sqs_max_messages_per_poll";
         public static final String OptOutSqsVisibilityTimeoutProp = "optout_sqs_visibility_timeout";
         public static final String OptOutDeltaJobTimeoutSecondsProp = "optout_delta_job_timeout_seconds";

src/main/java/com/uid2/optout/Main.java

@@ -246,7 +246,7 @@ public void run(String[] args) throws IOException {
 
         List<Future> futs = new ArrayList<>();
 
-        // create optout cloud sync verticle
+        // create optout cloud sync verticle (reads from optout_s3_folder)
         OptOutCloudSync cs = new OptOutCloudSync(this.config, true);
         CloudSyncVerticle cloudSyncVerticle = new CloudSyncVerticle("optout", this.fsOptOut, this.fsLocal, cs, this.config);
 
@@ -268,8 +268,21 @@ public void run(String[] args) throws IOException {
             OptOutLogProducer logProducer = new OptOutLogProducer(this.config, eventUpload, eventUpload);
             futs.add(this.deploySingleInstance(logProducer));
 
+            // create cloud sync for legacy producer uploads
+            // if optout_legacy_producer_s3_folder is set, old producer writes to that folder instead of optout_s3_folder
+            String legacyFolder = this.config.getString(Const.Config.OptOutLegacyProducerS3FolderProp);
+            OptOutCloudSync legacyProducerCs;
+            if (legacyFolder != null) {
+                JsonObject legacyConfig = new JsonObject().mergeIn(this.config)
+                    .put(Const.Config.OptOutS3FolderProp, legacyFolder);
+                legacyProducerCs = new OptOutCloudSync(legacyConfig, true);
+                LOGGER.info("legacy producer will write to separate folder: {}", legacyFolder);
+            } else {
+                legacyProducerCs = cs;
+            }
+
             // upload last delta produced and potentially not uploaded yet
-            futs.add((this.uploadLastDelta(cs, logProducer, cloudSyncVerticle.eventUpload(), cloudSyncVerticle.eventRefresh())));
+            futs.add((this.uploadLastDelta(legacyProducerCs, logProducer, cloudSyncVerticle.eventUpload(), cloudSyncVerticle.eventRefresh())));
         }
 
         // deploy sqs producer if enabled