Fix tests

PostHog · Sep 19, 2023 · 7a3f278 · 7a3f278
1 parent 4be4f5c
commit 7a3f278
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/plugin-server/src/main/ingestion-queues/session-recording/services/realtime-manager.ts b/plugin-server/src/main/ingestion-queues/session-recording/services/realtime-manager.ts
@@ -63,6 +63,7 @@ export class RealtimeManager extends EventEmitter {
 
     public async unsubscribe(): Promise<void> {
         await this.pubsubRedis?.unsubscribe(Keys.realtimeSubscriptions())
+        await this.pubsubRedis?.quit()
     }
 
     private async run<T>(description: string, fn: (client: Redis) => Promise<T>): Promise<T | null> {

diff --git a/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer-v2.ts b/plugin-server/src/main/ingestion-queues/session-recording/session-recordings-consumer-v2.ts
@@ -546,7 +546,10 @@ export class SessionRecordingIngesterV2 {
         if (this.serverConfig.SESSION_RECORDING_PARTITION_REVOKE_OPTIMIZATION) {
             status.info('🔁', `blob_ingester_consumer - flushing ${sessionsToDrop.length} sessions on revoke...`)
             await Promise.allSettled(
-                sessionsToDrop.map(([_, sessionManager]) => sessionManager.flush('partition_shutdown'))
+                sessionsToDrop
+                    .map(([_, x]) => x)
+                    .sort((x) => x.buffer.oldestKafkaTimestamp ?? Infinity)
+                    .map((x) => x.flush('partition_shutdown'))
             )
         }