CR changes

snowflakedb · Oct 17, 2024 · 0ef457a · 0ef457a
1 parent bba1a05
commit 0ef457a
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 21 deletions.
diff --git a/.../com/snowflake/kafka/connector/streaming/iceberg/IcebergIngestionNoSchemaEvolutionIT.java b/.../com/snowflake/kafka/connector/streaming/iceberg/IcebergIngestionNoSchemaEvolutionIT.java
@@ -11,6 +11,18 @@
 
 public class IcebergIngestionNoSchemaEvolutionIT extends IcebergIngestionIT {
 
+  private static final String RECORD_CONTENT_OBJECT_SCHEMA =
+      "object("
+          + "id_int8 NUMBER(10,0),"
+          + "id_int16 NUMBER(10,0),"
+          + "id_int32 NUMBER(10,0),"
+          + "id_int64 NUMBER(19,0),"
+          + "description VARCHAR(16777216),"
+          + "rating_float32 FLOAT,"
+          + "rating_float64 FLOAT,"
+          + "approval BOOLEAN"
+          + ")";
+
   @Override
   protected Boolean isSchemaEvolutionEnabled() {
     return false;
@@ -29,17 +41,11 @@ protected void createIcebergTable() {
             + RECORD_CONTENT_OBJECT_SCHEMA);
   }
 
-  private static final String RECORD_CONTENT_OBJECT_SCHEMA =
-      "object("
-          + "id_int8 NUMBER(10,0),"
-          + "id_int16 NUMBER(10,0),"
-          + "id_int32 NUMBER(10,0),"
-          + "id_int64 NUMBER(19,0),"
-          + "description VARCHAR(16777216),"
-          + "rating_float32 FLOAT,"
-          + "rating_float64 FLOAT,"
-          + "approval BOOLEAN"
-          + ")";
+  private static Stream<Arguments> prepareData() {
+    return Stream.of(
+        Arguments.of("Primitive JSON with schema", primitiveJsonWithSchema, true),
+        Arguments.of("Primitive JSON without schema", primitiveJson, false));
+  }
 
   @ParameterizedTest(name = "{0}")
   @MethodSource("prepareData")
@@ -53,10 +59,4 @@ void shouldInsertRecords(String description, String message, boolean withSchema)
     service.insert(Collections.singletonList(createKafkaRecord(message, 2, withSchema)));
     waitForOffset(3);
   }
-
-  private static Stream<Arguments> prepareData() {
-    return Stream.of(
-        Arguments.of("Primitive JSON with schema", primitiveJsonWithSchema, true),
-        Arguments.of("Primitive JSON without schema", primitiveJson, false));
-  }
 }
diff --git a/...va/com/snowflake/kafka/connector/streaming/iceberg/IcebergIngestionSchemaEvolutionIT.java b/...va/com/snowflake/kafka/connector/streaming/iceberg/IcebergIngestionSchemaEvolutionIT.java
@@ -34,8 +34,10 @@ void shouldEvolveSchemaAndInsertRecords(
       throws Exception {
     // start off with just one column
     List<DescribeTableRow> rows = describeTable(tableName);
-    assertThat(rows.size()).isEqualTo(1);
-    assertThat(rows.get(0).getColumn()).isEqualTo(Utils.TABLE_COLUMN_METADATA);
+    assertThat(rows)
+        .hasSize(1)
+        .extracting(DescribeTableRow::getColumn)
+        .contains(Utils.TABLE_COLUMN_METADATA);
 
     SinkRecord record = createKafkaRecord(message, 0, withSchema);
     service.insert(Collections.singletonList(record));
@@ -63,8 +65,7 @@ void shouldEvolveSchemaAndInsertRecords(
     service.insert(Collections.singletonList(createKafkaRecord(simpleRecordJson, 2, false)));
 
     rows = describeTable(tableName);
-    assertThat(rows.size()).isEqualTo(10);
-    assertThat(rows).contains(new DescribeTableRow("SIMPLE", "VARCHAR(16777216)"));
+    assertThat(rows).hasSize(10).contains(new DescribeTableRow("SIMPLE", "VARCHAR(16777216)"));
 
     // reinsert record with extra field
     service.insert(Collections.singletonList(createKafkaRecord(simpleRecordJson, 2, false)));