fix(query-orchestrator): streams cluster (#6048)

cube-js · Jan 23, 2023 · c5b6702 · c5b6702
1 parent 69143d5
commit c5b6702
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 1 deletion.
diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.js b/packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.js
@@ -782,7 +782,7 @@ export class QueryQueue {
           let target;
           switch (handler) {
             case 'stream':
-              target = this.getQueryStream(activeKeys[0]);
+              target = this.getQueryStream(this.redisHash(queryKey));
               await this.queryTimeout(this.queryHandlers.stream(query.query, target));
               break;
             default:
@@ -897,6 +897,11 @@ export class QueryQueue {
           activated,
           queryExists: !!query
         });
+        // closing stream
+        if (query?.queryHandler === 'stream') {
+          const stream = this.getQueryStream(this.redisHash(queryKey));
+          stream.destroy();
+        }
         const currentProcessingId = await queueConnection.freeProcessingLock(queryKey, processingId, activated);
         if (currentProcessingId) {
           this.logger('Skipping free processing lock', {

diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryStream.ts b/packages/cubejs-query-orchestrator/src/orchestrator/QueryStream.ts
@@ -35,6 +35,9 @@ export class QueryStream extends stream.Transform {
     this.queryKey = key;
     this.maps = maps;
     this.aliasNameToMember = aliasNameToMember;
+    if (!this.aliasNameToMember) {
+      this.emit('error', 'The QueryStream `aliasNameToMember` property is missed.');
+    }
   }
 
   /**