Fixed issues identified while testing.

Author: sagarkapare

delta-plugins-common/src/main/java/io/cdap/delta/plugin/common/Records.java

@@ -304,7 +304,8 @@ public static Schema convert(org.apache.kafka.connect.data.Schema schema) {
         for (Field field : fields) {
           cdapFields.add(Schema.Field.of(field.name(), convert(field.schema())));
         }
-        converted = Schema.recordOf(schema.name(), cdapFields);
+
+        converted = Schema.recordOf(schema.name() == null ? "dummy.schema.name" : schema.name(), cdapFields);
         break;
       default:
         // should never happen, all values are listed above

mysql-delta-plugins/src/main/java/io/debezium/connector/mysql/MySqlJdbcContext.java

@@ -92,9 +92,7 @@ public MySqlJdbcContext(MySqlConnectorConfig config) {
 
     jdbcConfig = jdbcConfigBuilder.build();
     String driverClassName = jdbcConfig.getString(MySqlConnectorConfig.JDBC_DRIVER);
-    this.jdbc = new JdbcConnection(jdbcConfig,
-                                   JdbcConnection.patternBasedFactory(MYSQL_CONNECTION_URL, driverClassName,
-                                                                      getClass().getClassLoader()));
+    this.jdbc = new JdbcConnection(jdbcConfig, connectionFactory);
   }
 
   public Configuration config() {