Adding validation for psc id for running multiple instances

pravega · Mar 4, 2024 · ea39c04 · ea39c04
1 parent 35a1487
commit ea39c04
Show file tree

Hide file tree

Showing 6 changed files with 20 additions and 7 deletions.
diff --git a/pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/file/FileConfig.java b/pravega-sensor-collector/src/main/java/io/pravega/sensor/collector/file/FileConfig.java
@@ -33,11 +33,13 @@ public class FileConfig {
 
     public final boolean enableLargeEvent;
 
+    public final String pscId;
+
 
     public FileConfig(String stateDatabaseFileName, String fileSpec, String fileExtension, String routingKey,
                       String streamName, String eventTemplateStr, int maxRecordsPerEvent, boolean enableDeleteCompletedFiles,
                       boolean exactlyOnce, double transactionTimeoutMinutes, long minTimeInMillisToUpdateFile, String fileType,
-                      boolean enableLargeEvent) {
+                      boolean enableLargeEvent, String pscId) {
         this.stateDatabaseFileName = stateDatabaseFileName;
         this.fileSpec = fileSpec;
         this.fileExtension = fileExtension;
@@ -51,6 +53,7 @@ public FileConfig(String stateDatabaseFileName, String fileSpec, String fileExte
         this.minTimeInMillisToUpdateFile = minTimeInMillisToUpdateFile;
         this.fileType = fileType;
         this.enableLargeEvent = enableLargeEvent;
+        this.pscId = pscId;
     }
 
     @Override
@@ -69,6 +72,7 @@ public String toString() {
                 ", transactionTimeoutMinutes=" + transactionTimeoutMinutes +
                 ", minTimeInMillisToUpdateFile=" + minTimeInMillisToUpdateFile +
                 ", enableLargeEvent=" + enableLargeEvent +
+                ", pscId=" + pscId +
                 '}';
     }
 }
diff --git a/...ga-sensor-collector/src/main/java/io/pravega/sensor/collector/file/FileIngestService.java b/...ga-sensor-collector/src/main/java/io/pravega/sensor/collector/file/FileIngestService.java
@@ -52,6 +52,8 @@ public abstract class FileIngestService extends DeviceDriver {
     private static final int DEFAULT_SAMPLES_PER_EVENT_KEY = 100;
 
     private static final int DEFAULT_INTERVAL_MS_KEY = 10000;
+
+    private static final String PSC_ID = "PSC_ID";
     private final FileProcessor processor;
     private final MetricPublisher metricPublisher;
     private final ScheduledExecutorService executor;
@@ -75,7 +77,8 @@ public FileIngestService(DeviceDriverConfig config) {
                 getTransactionTimeoutMinutes(),
                 getMinTimeInMillisToUpdateFile(),
                 config.getClassName(),
-                getLargeEventEnable());
+                getLargeEventEnable(),
+                getPscId());
         LOG.info("File Ingest Config: {}", fileSequenceConfig);
         final String scopeName = getScopeName();
         LOG.info("Scope: {}", scopeName);
@@ -88,6 +91,10 @@ public FileIngestService(DeviceDriverConfig config) {
         executor = Executors.newScheduledThreadPool(1, namedThreadFactory);
     }
 
+    private String getPscId() {
+        return getProperty(PSC_ID);
+    }
+
     String getFileSpec() {
         return getProperty(FILE_SPEC_KEY);
     }

diff --git a/...or-collector/src/test/java/io/pravega/sensor/collector/file/FileProcessorFactoryTest.java b/...or-collector/src/test/java/io/pravega/sensor/collector/file/FileProcessorFactoryTest.java
@@ -46,7 +46,7 @@ public void createRAWFileProcessorTest() throws Exception {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig(stateDatabaseFileName,"/opt/pravega-sensor-collector/Files/A","parquet","key12",
                 "stream1","{}",10, false,
-                true,20.0, 5000l,"RawFileIngestService", true);
+                true,20.0, 5000l,"RawFileIngestService", true, "psc1");
         FileProcessor rawFileProcessor = FileProcessorFactory.createFileSequenceProcessor(config,state,writer,transactionCoordinator,"writerId");
 
         Assertions.assertTrue(rawFileProcessor instanceof RawFileProcessor);
@@ -61,7 +61,7 @@ public void createCSVFileProcessorTest() throws Exception {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig(stateDatabaseFileName,"/opt/pravega-sensor-collector/Files/A","parquet","key12",
                 "stream1","{}",10, false,
-                true,20.0, 5000L,"CsvFileIngestService", false);
+                true,20.0, 5000L,"CsvFileIngestService", false, "psc1");
         FileProcessor csvFileProcessor = FileProcessorFactory.createFileSequenceProcessor(config,state,writer,transactionCoordinator,"writerId");
 
         Assertions.assertTrue(csvFileProcessor instanceof CsvFileSequenceProcessor);
@@ -76,7 +76,7 @@ public void createParquetFileProcessorTest() throws Exception {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig(stateDatabaseFileName,"/opt/pravega-sensor-collector/Files/A","parquet","key12",
                 "stream1","{}",10, false,
-                true,20.0, 5000L,"ParquetFileIngestService", false);
+                true,20.0, 5000L,"ParquetFileIngestService", false, "psc1");
         FileProcessor parquetFileProcessor = FileProcessorFactory.createFileSequenceProcessor(config,state,writer,transactionCoordinator,"writerId");
 
         Assertions.assertTrue(parquetFileProcessor instanceof ParquetFileProcessor);

diff --git a/...a-sensor-collector/src/test/java/io/pravega/sensor/collector/file/FileProcessorTests.java b/...a-sensor-collector/src/test/java/io/pravega/sensor/collector/file/FileProcessorTests.java
@@ -71,7 +71,7 @@ protected void setup() {
         String stateDatabaseFileName = ":memory:";
         config = new FileConfig("./psc.db","/opt/pravega-sensor-collector/Files/A","parquet","key12",
                 "stream1","{}",10, true,
-                true,20.0, 5000,"RawFileIngestService", true);
+                true,20.0, 5000,"RawFileIngestService", true, "pscId");
     }
 
     @Test
@@ -214,7 +214,7 @@ public void testCreateRawFileProcessorWithNullTransactionCordinator() {
     public void testCreateRawFileProcessorWithNullStateDatabaseFilenameInConfig() {
         FileConfig newConfig = new FileConfig(null,"/opt/pravega-sensor-collector/Files/A","parquet","key12",
                 "stream1","{}",10, false,
-                true,20.0, 5000,"RawFileIngestService", true);
+                true,20.0, 5000,"RawFileIngestService", true, "pscId");
         Exception exception = Assert.assertThrows(NullPointerException.class, () -> new RawFileProcessor(newConfig, state, transactionalEventWriter, transactionCoordinator, "test"));
         Assert.assertTrue("config.stateDatabaseFileName".equals(exception.getMessage()));
     }

diff --git a/pravega-sensor-collector/src/test/resources/RawFileIngest-integration-test.properties b/pravega-sensor-collector/src/test/resources/RawFileIngest-integration-test.properties
@@ -22,3 +22,4 @@ PRAVEGA_SENSOR_COLLECTOR_RAW1_TRANSACTION_TIMEOUT_MINUTES=2.0
 PRAVEGA_SENSOR_COLLECTOR_RAW1_CREATE_SCOPE=true
 PRAVEGA_SENSOR_COLLECTOR_RAW1_MIN_TIME_IN_MILLIS_TO_UPDATE_FILE=5000
 PRAVEGA_SENSOR_COLLECTOR_RAW1_DELETE_COMPLETED_FILES_INTERVAL_IN_SECONDS=15
+PRAVEGA_SENSOR_COLLECTOR_RAW1_PSC_ID=test
diff --git a/pravega-sensor-collector/src/test/resources/RawFileIngestService.properties b/pravega-sensor-collector/src/test/resources/RawFileIngestService.properties
@@ -24,3 +24,4 @@ PRAVEGA_SENSOR_COLLECTOR_RAW1_TRANSACTION_TIMEOUT_MINUTES=1.0
 PRAVEGA_SENSOR_COLLECTOR_RAW1_CREATE_SCOPE=true
 PRAVEGA_SENSOR_COLLECTOR_RAW1_MIN_TIME_IN_MILLIS_TO_UPDATE_FILE=5000
 PRAVEGA_SENSOR_COLLECTOR_RAW1_DELETE_COMPLETED_FILES_INTERVAL_IN_SECONDS=15
+PRAVEGA_SENSOR_COLLECTOR_RAW1_PSC_ID=test