fix: failure to apply default naming strategy when topic has not the expected prefix

Author: teoincontatto

stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/target/migration/StreamMigrationTableNamingStrategy.java

@@ -20,8 +20,10 @@ public static void setTopicPrefix(String topicPrefix) {
   @Override
   public String resolveTableName(JdbcSinkConnectorConfig config, SinkRecord record) {
     String table = super.resolveTableName(config, record);
-    final String originalName = record.topic().substring(StreamMigrationTableNamingStrategy.topicPrefix.length());
-    table = table.replace("${original}", originalName);
+    if (record.topic().startsWith(StreamMigrationTableNamingStrategy.topicPrefix)) {
+      final String originalName = record.topic().substring(StreamMigrationTableNamingStrategy.topicPrefix.length());
+      table = table.replace("${original}", originalName);
+    }
     return table;
   }