[CYB-185] indexing doesnt start when two sources map to the same topic (

#64)
cloudera · May 16, 2024 · 46f8087 · 46f8087
1 parent aa113af
commit 46f8087
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 2 deletions.
diff --git a/...ing-hive/src/main/java/com/cloudera/cyber/indexing/hive/tableapi/TableApiAbstractJob.java b/...ing-hive/src/main/java/com/cloudera/cyber/indexing/hive/tableapi/TableApiAbstractJob.java
@@ -334,13 +334,17 @@ private void configure(StreamTableEnvironment tableEnv) {
 
   protected final String buildInsertSql(String topic, MappingDto mappingDto, ResolvedSchema tableSchema) {
     return String.join("\n",
-        getInsertSqlPrefix() + " " + mappingDto.getTableName() + "(" + getInsertColumns(mappingDto) + ") "
+        getInsertSqlPrefix() + " " + getTableName(topic, mappingDto) + "(" + getInsertColumns(mappingDto) + ") "
         + getInsertSqlSuffix(),
         " SELECT " + getFromColumns(mappingDto, tableSchema),
         " from " + KAFKA_TABLE,
         String.format(" where `source`='%s'", topic));
   }
 
+  protected String getTableName(String source, MappingDto mappingDto) {
+    return mappingDto.getTableName();
+  }
+
   protected String getInsertSqlPrefix() {
     return "INSERT INTO ";
   }

diff --git a/...g-hive/src/main/java/com/cloudera/cyber/indexing/hive/tableapi/impl/TableApiKafkaJob.java b/...g-hive/src/main/java/com/cloudera/cyber/indexing/hive/tableapi/impl/TableApiKafkaJob.java
@@ -51,7 +51,7 @@ protected void executeInsert(StreamTableEnvironment tableEnv, Map<String, Mappin
             mappingDto.getTableName(), "indexing-job", params);
 
         //read from view and write to kafka sink
-        final Table table = tableEnv.from(mappingDto.getTableName());
+        final Table table = tableEnv.from(getTableName(topic, mappingDto));
         final String schemaString = AvroSchemaUtil.convertToAvro(tablesConfig.get(mappingDto.getTableName()))
             .toString();
 
@@ -98,6 +98,11 @@ protected FormatDescriptor getFormatDescriptor() {
     return null;
   }
 
+  @Override
+  protected String getTableName(String source, MappingDto mappingDto) {
+    return source.concat("_tmpview");
+  }
+
   @Override
   protected String getInsertSqlPrefix() {
     return "CREATE TEMPORARY VIEW ";