SNOW-1728000 Creating Iceberg table in e2e tests

test/rest_request_template/iceberg_json_aws.json

@@ -0,0 +1,30 @@
+{
+  "name": "SNOWFLAKE_CONNECTOR_NAME",
+  "config": {
+    "connector.class": "com.snowflake.kafka.connector.SnowflakeSinkConnector",
+    "topics": "SNOWFLAKE_TEST_TOPIC",
+    "tasks.max": "1",
+    "buffer.flush.time": "10",
+    "buffer.count.records": "100",
+    "buffer.size.bytes": "5000000",
+    "snowflake.url.name": "SNOWFLAKE_HOST",
+    "snowflake.user.name": "SNOWFLAKE_USER",
+    "snowflake.private.key": "SNOWFLAKE_PRIVATE_KEY",
+    "snowflake.database.name": "SNOWFLAKE_DATABASE",
+    "snowflake.schema.name": "SNOWFLAKE_SCHEMA",
+    "snowflake.role.name": "SNOWFLAKE_ROLE",
+    "snowflake.ingestion.method": "SNOWPIPE_STREAMING",
+    "snowflake.streaming.enable.single.buffer": true,
+    "snowflake.streaming.closeChannelsInParallel.enabled": true,
+    "snowflake.enable.schematization": "false",
+    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
+    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+    "value.converter.schemas.enable": "false",
+    "jmx": "true",
+    "errors.tolerance": "all",
+    "errors.log.enable": true,
+    "errors.deadletterqueue.topic.name": "DLQ_TOPIC",
+    "errors.deadletterqueue.topic.replication.factor": 1,
+    "snowflake.streaming.iceberg.enabled": true
+  }
+}

test/test_executor.py

@@ -8,6 +8,7 @@ class TestExecutor:
     def execute(self, testSuitList, driver, nameSalt, round=1):
         try:
             for test in testSuitList:
+                test.setup()
                 driver.createConnector(test.getConfigFileName(), nameSalt)
 
             driver.startConnectorWaitTime()

test/test_suit/base_e2e.py

@@ -0,0 +1,3 @@
+class BaseE2eTest:
+    def setup(self):
+        pass

test/test_suit/iceberg_json_aws.py

@@ -0,0 +1,41 @@
+import datetime
+
+from test_suit.test_utils import RetryableError, NonRetryableError
+import json
+from time import sleep
+from test_suit.base_e2e import BaseE2eTest
+
+
+class TestIcebergJsonAws(BaseE2eTest):
+    def __init__(self, driver, nameSalt: str):
+        self.driver = driver
+        self.fileName = "iceberg_json_aws"
+        self.topic = self.fileName + nameSalt
+
+    def getConfigFileName(self):
+        return self.fileName + ".json"
+
+    def setup(self):
+        self.driver.create_iceberg_table_with_content(
+            table_name=self.topic,
+            external_volume="kafka_push_e2e_volume_aws",  # volume created manually
+        )
+
+    def send(self):
+        msg = json.dumps(
+            {
+                "id": 1,
+                "body_temperature": 36.6,
+                "name": "Steve",
+                "approved_coffee_types": ["Espresso"],
+                "animals_possessed": {"dogs": True, "cats": False},
+            }
+        )
+
+    def verify(self, round):
+        res = self.driver.select_number_of_records(self.topic)
+        print("Count records in table {}={}".format(self.topic, str(res)))
+
+    def clean(self):
+        self.driver.drop_iceberg_table(self.topic)
+        return

test/test_suit/resilience_tests/test_kc_delete_create.py

@@ -3,13 +3,14 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/2
 # deletes the connector
 # creates the connector
 # sends data 2/2
 # verifies that 2 rounds of data were ingested
-class TestKcDeleteCreate:
+class TestKcDeleteCreate(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -47,8 +48,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_delete_create_chaos.py

@@ -3,14 +3,15 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/3
 # deletes the connector
 # sends data 2/3
 # creates the connector
 # sends data 3/3
 # verifies that 3 rounds of data were ingested
-class TestKcDeleteCreateChaos:
+class TestKcDeleteCreateChaos(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -50,8 +51,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_delete_resume.py

@@ -3,13 +3,14 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/2
 # deletes the connector
 # resumes the connector (will not work, because connector was deleted)
 # sends data 2/2
 # verifies that 1 round of data was ingested
-class TestKcDeleteResume:
+class TestKcDeleteResume(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -48,8 +49,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_delete_resume_chaos.py

@@ -3,14 +3,15 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/3
 # deletes the connector
 # sends data 2/3
 # resumes the connector (will not work, because connector was deleted)
 # sends data 2/3
 # verifies that 2-3 rounds of data was ingested, since some round 2 data may have been ingested prior to connector deletion
-class TestKcDeleteResumeChaos:
+class TestKcDeleteResumeChaos(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -53,8 +54,7 @@ def verify(self, round):
         # since the pressure is applied during deletion, some of the data may be ingested, so look for a range
         goalCountUpper = self.recordNum * self.expectedsends
         goalCountLower = self.recordNum * (self.expectedsends - 1)
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count between: {} - {}".format(self.topic, str(res), str(goalCountLower), str(goalCountUpper)))
 

test/test_suit/resilience_tests/test_kc_pause_create.py

@@ -3,13 +3,14 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/2
 # pauses the connector
 # creates the connector
 # sends data 2/2
 # verifies that 2 rounds of data were ingested
-class TestKcPauseCreate:
+class TestKcPauseCreate(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -47,8 +48,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_pause_create_chaos.py

@@ -3,14 +3,15 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/3
 # pauses the connector
 # sends data 2/3
 # creates the connector
 # sends data 3/3
 # verifies that 3 rounds of data were ingested
-class TestKcPauseCreateChaos:
+class TestKcPauseCreateChaos(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -50,8 +51,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_pause_resume.py

@@ -3,13 +3,14 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/2
 # pauses the connector
 # resumes the connector
 # sends data 2/2
 # verifies that 2 rounds of data were ingested
-class TestKcPauseResume:
+class TestKcPauseResume(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -47,8 +48,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_pause_resume_chaos.py

@@ -3,14 +3,15 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/3
 # pauses the connector
 # sends data 2/3
 # resumes the connector
 # sends data 3/3
 # verifies that 3 rounds of data were ingested
-class TestKcPauseResumeChaos:
+class TestKcPauseResumeChaos(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -50,8 +51,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_recreate.py

@@ -3,13 +3,14 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/2
 # creates the connector 1/2
 # creates the connector 2/2
 # sends data 2/2
 # verifies that 2 rounds of data were ingested
-class TestKcRecreate:
+class TestKcRecreate(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -47,8 +48,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_recreate_chaos.py

@@ -3,14 +3,15 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/3
 # creates the connector 1/2
 # sends data 2/3
 # creates the connector 2/2
 # sends data 3/3
 # verifies that 3 rounds of data were ingested
-class TestKcRecreateChaos:
+class TestKcRecreateChaos(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -48,8 +49,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))
 

test/test_suit/resilience_tests/test_kc_restart.py

@@ -3,6 +3,7 @@
 from test_suit.test_utils import RetryableError, NonRetryableError, ResetAndRetry
 import json
 from time import sleep
+from test_suit.base_e2e import BaseE2eTest
 
 # sends data 1/3
 # restarts the connector
@@ -13,7 +14,7 @@
 
 # a pressure test cannot be created with this, since restart is just one method
 # restarting the connector and restarting the connector and all its tasks should technically be separate tests, but should be fine to group here
-class TestKcRestart:
+class TestKcRestart(BaseE2eTest):
     def __init__(self, driver, nameSalt):
         self.driver = driver
         self.nameSalt = nameSalt
@@ -53,8 +54,7 @@ def send(self):
     def verify(self, round):
         # verify record count
         goalCount = self.recordNum * self.expectedsends
-        res = self.driver.snowflake_conn.cursor().execute(
-            "SELECT count(*) FROM {}".format(self.topic)).fetchone()[0]
+        res = self.driver.select_number_of_records(self.topic)
 
         print("Count records in table {}={}. Goal record count: {}".format(self.topic, str(res), str(goalCount)))