Cleanup YAML from beta testing (#174)

Author: george-zubrienko

integration-tests.env

@@ -1,4 +1,4 @@
-STREAMCONTEXT__SPEC='{ "backfillJobTemplateRef": { "apiGroup": "streaming.sneaksanddata.com", "kind": "StreamingJobTemplate", "name": "arcane-stream-microsoft-synapse-link-large-job" }, "groupingIntervalSeconds": 1, "httpClientMaxRetries": 3, "httpClientRetryDelaySeconds": 1, "jobTemplateRef": { "apiGroup": "streaming.sneaksanddata.com", "kind": "StreamingJobTemplate", "name": "arcane-stream-microsoft-synapse-link-standard-job" }, "lookBackInterval": 21000, "tableProperties": { "partitionExpressions": [], "format": "PARQUET", "sortedBy": [], "parquetBloomFilterColumns": [] }, "rowsPerGroup": 10000, "sinkSettings": { "archiveTableName": "iceberg.test.archive_test", "optimizeSettings": { "batchThreshold": 60, "fileSizeThreshold": "512MB" }, "orphanFilesExpirationSettings": { "batchThreshold": 60, "retentionThreshold": "6h" }, "snapshotExpirationSettings": { "batchThreshold": 60, "retentionThreshold": "6h" }, "analyzeSettings" : { "batchThreshold": 60, "includedColumns": [] }, "targetTableName": "iceberg.test.test" }, "sourceSettings": { "baseLocation": "abfss://cdm-e2e@devstoreaccount1.dfs.core.windows.net/", "name": "dimensionattributelevelvalue", "changeCaptureIntervalSeconds": 5 }, "stagingDataSettings": { "maxRowsPerFile": 10000, "catalog": { "catalogName": "iceberg", "catalogUri": "http://localhost:20001/catalog", "schemaName": "test", "warehouse": "demo" },  "tableNamePrefix": "staging_dimensionattributelevelvalue" }, "fieldSelectionRule": { "ruleType": "all", "fields": [] }, "backfillBehavior": "overwrite", "backfillStartDate": "2025-03-04T07.00.00Z" }'
+STREAMCONTEXT__SPEC='{ "backfillJobTemplateRef": { "apiGroup": "streaming.sneaksanddata.com", "kind": "StreamingJobTemplate", "name": "arcane-stream-microsoft-synapse-link-large-job" }, "groupingIntervalSeconds": 1, "httpClientMaxRetries": 3, "httpClientRetryDelaySeconds": 1, "jobTemplateRef": { "apiGroup": "streaming.sneaksanddata.com", "kind": "StreamingJobTemplate", "name": "arcane-stream-microsoft-synapse-link-standard-job" }, "lookBackInterval": 21000, "tableProperties": { "partitionExpressions": [], "format": "PARQUET", "sortedBy": [], "parquetBloomFilterColumns": [] }, "rowsPerGroup": 10000, "sinkSettings": { "archiveTableName": "iceberg.test.archive_test", "optimizeSettings": { "batchThreshold": 60, "fileSizeThreshold": "512MB" }, "orphanFilesExpirationSettings": { "batchThreshold": 60, "retentionThreshold": "6h" }, "snapshotExpirationSettings": { "batchThreshold": 60, "retentionThreshold": "6h" }, "analyzeSettings" : { "batchThreshold": 60, "includedColumns": [] }, "targetTableName": "iceberg.test.test", "sinkCatalogSettings": {"namespace": "test", "warehouse": "demo", "catalogUri": "http://localhost:20001/catalog" } }, "sourceSettings": { "baseLocation": "abfss://cdm-e2e@devstoreaccount1.dfs.core.windows.net/", "name": "dimensionattributelevelvalue", "changeCaptureIntervalSeconds": 5 }, "stagingDataSettings": { "maxRowsPerFile": 10000, "catalog": { "catalogName": "iceberg", "catalogUri": "http://localhost:20001/catalog", "schemaName": "test", "warehouse": "demo" },  "tableNamePrefix": "staging_dimensionattributelevelvalue" }, "fieldSelectionRule": { "ruleType": "all", "fields": [] }, "backfillBehavior": "overwrite", "backfillStartDate": "2025-03-04T07.00.00Z" }'
 STREAMCONTEXT__STREAM_ID=test
 STREAMCONTEXT__STREAM_KIND=MicrosoftSynapseStream
 ARCANE_FRAMEWORK__S3_CATALOG_ACCESS_KEY_ID=minioadmin
@@ -15,6 +15,3 @@ ARCANE_FRAMEWORK__STORAGE_CONTAINER=cdm-e2e
 ARCANE_FRAMEWORK__STORAGE_ENDPOINT=http://localhost:10001/devstoreaccount1
 ARCANE_FRAMEWORK__S3_CATALOG_AUTH_SESSION_TIMEOUT_MILLIS=2000
 ARCANE_FRAMEWORK__MERGE_SERVICE_CONNECTION_URI=jdbc:trino://localhost:8080/iceberg/test?user=test
-ARCANE_FRAMEWORK__ICEBERG_SINK_NAMESPACE=test
-ARCANE_FRAMEWORK__ICEBERG_SINK_WAREHOUSE=demo
-ARCANE_FRAMEWORK__ICEBERG_SINK_CATALOG_URI=http://localhost:20001/catalog

src/main/scala/models/app/MicrosoftSynapseLinkStreamContext.scala

@@ -170,14 +170,9 @@ case class MicrosoftSynapseLinkStreamContext(spec: StreamSpec)
     sys.env.getOrElse("ARCANE_FRAMEWORK__METRICS_PUBLISHER_INTERVAL_MILLIS", "100").toInt
   )
   override val icebergSinkSettings: IcebergSinkSettings = new IcebergSinkSettings {
-    override val namespace: String =
-      sys.env("ARCANE_FRAMEWORK__ICEBERG_SINK_NAMESPACE") // spec.sinkSettings.sinkCatalogSettings.namespace.getOrElse()
-    override val warehouse: String =
-      sys.env("ARCANE_FRAMEWORK__ICEBERG_SINK_WAREHOUSE") // spec.sinkSettings.sinkCatalogSettings.warehouse.getOrElse()
-    override val catalogUri: String =
-      sys.env(
-        "ARCANE_FRAMEWORK__ICEBERG_SINK_CATALOG_URI"
-      ) // spec.sinkSettings.sinkCatalogSettings.catalogUri.getOrElse()
+    override val namespace: String                         = spec.sinkSettings.sinkCatalogSettings.namespace
+    override val warehouse: String                         = spec.sinkSettings.sinkCatalogSettings.warehouse
+    override val catalogUri: String                        = spec.sinkSettings.sinkCatalogSettings.catalogUri
     override val additionalProperties: Map[String, String] = IcebergCatalogCredential.oAuth2Properties
   }
 

src/main/scala/models/app/contracts/StreamSpec.scala

@@ -37,9 +37,9 @@ case class OrphanFilesExpirationSettings(batchThreshold: Int, retentionThreshold
 case class AnalyzeSettings(batchThreshold: Int, includedColumns: Seq[String]) derives ReadWriter
 
 case class IcebergSinkSettings(
-    namespace: Option[String] = None,
-    warehouse: Option[String] = None,
-    catalogUri: Option[String] = None
+    namespace: String,
+    warehouse: String,
+    catalogUri: String
 ) derives ReadWriter
 
 /** The configuration of Iceberg sink.
@@ -50,7 +50,7 @@ case class SinkSettings(
     snapshotExpirationSettings: SnapshotExpirationSettingsSpec,
     orphanFilesExpirationSettings: OrphanFilesExpirationSettings,
     analyzeSettings: AnalyzeSettings,
-    sinkCatalogSettings: Option[IcebergSinkSettings] = None
+    sinkCatalogSettings: IcebergSinkSettings
 ) derives ReadWriter
 
 /** The configuration of the stream source.

src/test/scala/integration/StreamRunner.scala

@@ -53,7 +53,12 @@ object StreamRunner extends ZIOSpecDefault:
     |      "batchThreshold": 60,
     |      "includedColumns": []
     |    },
-    |    "targetTableName": "$targetTableName"
+    |    "targetTableName": "$targetTableName",
+    |    "sinkCatalogSettings": {
+    |      "namespace": "test",
+    |      "warehouse": "demo",
+    |      "catalogUri": "http://localhost:20001/catalog"
+    |    }
     |  },
     |  "sourceSettings": {
     |    "baseLocation": "abfss://cdm-e2e@devstoreaccount1.dfs.core.windows.net/",
@@ -138,8 +143,6 @@ object StreamRunner extends ZIOSpecDefault:
         )
 
         streamedWatermark <- Common.getWatermark(streamingStreamContext.targetTableFullName.split('.').last)
-
-      // TODO: verify watermarks
       yield assertTrue(result.isSuccess) implies assertTrue(backfilledCount == 5) implies assertTrue(
         backfilledWatermark.version == s"${formatter.format(startTime.minusHours(1))}Z"
       ) implies assertTrue(currentRows.size == 5 - 1 + 2) implies assertTrue(