Add postData retry with backoff. Timing profile logs commented out.

Author: callumstew

src/app/core/services/kafka/converters/healthkit-converter.service.ts

@@ -129,7 +129,7 @@ export class HealthkitConverterService extends ConverterService {
   convertToHealthkitRecordPaged$(
     kafkaValue,
     valueSchemaMetadata,
-    pageSize = 1000
+    pageSize = 3600
   ): Observable<any[]> {
     const data = kafkaValue.data
     const name = data.key

src/app/core/services/kafka/kafka.service.ts

@@ -30,7 +30,7 @@ import { AnalyticsService } from '../usage/analytics.service'
 import { CacheService } from './cache.service'
 import { SchemaService } from './schema.service'
 import { from, isObservable, Observable, of, Subject } from 'rxjs'
-import { concatMap } from 'rxjs/operators'
+import { mergeMap } from 'rxjs/operators'
 import { NotificationService } from '../notifications/notification.service'
 import { NotificationActionType } from 'src/app/shared/models/notification-handler'
 
@@ -52,9 +52,11 @@ interface SendFromCacheOptions {
 export class KafkaService {
   private static DEFAULT_TOPIC_CACHE_VALIDITY = 600_000 // 10 minutes
   private static BATCH_SIZE = 10
-  private static CONCURRENCY_LIMIT = 5
+  private static CONCURRENCY_LIMIT = 1
   private static SEND_ERROR_NOTIFICATION_THRESHOLD = 10
   private static HTTP_TIMEOUT = 120_000 // 2 minutes
+  private static HTTP_RETRY_MAX_ATTEMPTS = 3
+  private static HTTP_RETRY_BASE_DELAY_MS = 1000
 
   URI_topics: string = '/topics/'
   DEFAULT_KAFKA_AVSC = 'questionnaire'
@@ -280,7 +282,9 @@ export class KafkaService {
                 headers,
                 options
               )
-              if (sentRecordCount === 0) this.logger.log('Kafka record is empty, skipping sending')
+              if (sentRecordCount === 0) {
+                this.logger.log('Kafka record is empty, skipping sending')
+              }
               batchSuccessKeys.push(k)
             } catch (e) {
               if (e.message === 'Cache sending cancelled') {
@@ -384,7 +388,7 @@ export class KafkaService {
     return new Promise((resolve, reject) => {
       payloadStream
         .pipe(
-          concatMap(payload => {
+          mergeMap(payload => {
             if (this.cancelSending) {
               throw new Error('Cache sending cancelled')
             }
@@ -396,7 +400,7 @@ export class KafkaService {
                 () => payload.record.records.length
               )
             )
-          })
+          }, 1)
         )
         .subscribe({
           next: sentCount => {
@@ -423,13 +427,12 @@ export class KafkaService {
     return options.streamHealthkitPayloads !== false
   }
 
-  sendToKafka(topic, record, headers): Promise<any> {
-    const allRecords = record.records
+  async sendToKafka(topic, record, headers): Promise<any> {
     const jsonData = JSON.stringify(record)
     // Compress with pako and convert directly to base64
     const gzippedBytesB64 = fromByteArray(pako.gzip(jsonData))
 
-    return this.postData(
+    return await this.postData(
       gzippedBytesB64,
       topic,
       headers.set('Content-Encoding', DefaultCompressedContentEncoding),
@@ -486,7 +489,7 @@ export class KafkaService {
     return result
   }
 
-  postData(data: any, topic: string, headers: HttpHeaders, isBase64: boolean = false): Promise<any> {
+  async postData(data: any, topic: string, headers: HttpHeaders, isBase64: boolean = false): Promise<any> {
     const nativeHeaders = this.convertHeaders(headers)
     const request: any = {
       url: `${this.KAFKA_CLIENT_URL}${this.URI_topics}${topic}`,
@@ -503,27 +506,66 @@ export class KafkaService {
     request.connectionTimeout = KafkaService.HTTP_TIMEOUT
     request.readTimeout = KafkaService.HTTP_TIMEOUT
 
-    const requestPromise = Http.request(request)
-      .then(response => {
+    for (let attempt = 1; attempt <= KafkaService.HTTP_RETRY_MAX_ATTEMPTS; attempt++) {
+      // const httpStart = Date.now()
+      try {
+        const response = await Http.request(request)
+        // console.log('Kafka HTTP request time', {
+        //   topic,
+        //   status: response.status,
+        //   compressed: isBase64,
+        //   bytes: typeof data === 'string' ? data.length : undefined,
+        //   attempt,
+        //   milliseconds: Date.now() - httpStart
+        // })
         if (response.status < 200 || response.status >= 300) {
+          if (this.shouldRetryHttpStatus(response.status, attempt)) {
+            await this.delayHttpRetry(attempt, response.status, topic)
+            continue
+          }
           throw new HttpErrorResponse({
             error: response.data,
             status: response.status,
           })
         }
         return response
-      })
-      .catch(error => {
+      } catch (error) {
+        // console.log('Kafka HTTP request failed time', {
+        //   topic,
+        //   compressed: isBase64,
+        //   bytes: typeof data === 'string' ? data.length : undefined,
+        //   attempt,
+        //   milliseconds: Date.now() - httpStart
+        // })
         console.error('HTTP request failed:', error)
-
         if (this.cancelSending) {
           throw new Error('Request cancelled by user')
         }
+        if (this.shouldRetryHttpStatus(error?.status, attempt)) {
+          await this.delayHttpRetry(attempt, error.status, topic)
+          continue
+        }
+        throw error
+      }
+    }
 
-        throw new Error(`Failed to send data to Kafka: ${error.message}`)
-      })
+    throw new Error('Failed to send data to Kafka after retry attempts')
+  }
+
+  private shouldRetryHttpStatus(status: number, attempt: number): boolean {
+    return attempt < KafkaService.HTTP_RETRY_MAX_ATTEMPTS && (status === 500 || status === 503)
+  }
 
-    return requestPromise
+  private delayHttpRetry(attempt: number, status: number, topic: string): Promise<void> {
+    const delayMs = KafkaService.HTTP_RETRY_BASE_DELAY_MS * attempt
+    // console.log('Retrying Kafka HTTP request', {
+    //   topic,
+    //   status,
+    //   attempt,
+    //   nextAttempt: attempt + 1,
+    //   delayMs
+    // })
+    return new Promise(resolve => setTimeout(resolve, delayMs))
   }
 
   getAccessToken() {