DBZ-4272 Throw exception for non matching schema

debezium-core/src/main/java/io/debezium/util/ColumnUtils.java

@@ -11,9 +11,6 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.debezium.relational.Column;
 import io.debezium.relational.Table;
 
@@ -22,8 +19,6 @@
  */
 public class ColumnUtils {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(ColumnUtils.class);
-
     public static MappedColumns toMap(Table table) {
         Map<String, Column> sourceTableColumns = new HashMap<>();
         int greatestColumnPosition = 0;
@@ -54,14 +49,15 @@ public static ColumnArray toArray(ResultSet resultSet, Table table) throws SQLEx
                 for (int j = 0; j < metaData.getColumnCount(); j++) {
                     resultSetColumns[j] = metaData.getColumnName(j + 1);
                 }
-                LOGGER.warn(
-                        "Column '{}' not found in result set '{}' for table '{}', {}. This might be caused by DBZ-4350",
-                        columnName,
-                        String.join(", ", resultSetColumns),
-                        table.id(),
-                        table,
-                        new IllegalArgumentException("Columns in schema do not match result set"));
-                continue;
+                throw new IllegalArgumentException("Column '"
+                        + columnName
+                        + "' not found in result set '"
+                        + String.join(", ", resultSetColumns)
+                        + "' for table '"
+                        + table.id()
+                        + "', "
+                        + table
+                        + ". This might be caused by DBZ-4350");
             }
             greatestColumnPosition = greatestColumnPosition < columns[i].position()
                     ? columns[i].position()