[O2B-1412] Configure kafka to retry indefinitely

.eslintrc

@@ -85,7 +85,7 @@
         ],
         "function-paren-newline": [
             "error",
-            "multiline"
+            "multiline-arguments"
         ],
         "indent": [
             "error",

lib/application.js

@@ -10,6 +10,7 @@
  * granted to it by virtue of its status as an Intergovernmental Organization
  * or submit itself to any jurisdiction.
  */
+
 const database = require('./database');
 const { webUiServer } = require('./server');
 const { gRPCServer } = require('./server/index.js');
@@ -21,7 +22,7 @@ const { isInTestMode } = require('./utilities/env-utils.js');
 const { ScheduledProcessesManager } = require('./server/services/ScheduledProcessesManager.js');
 const { MonAlisaSynchronizer } = require('./server/externalServicesSynchronization/monalisa/MonAlisaSynchronizer');
 const { LogManager } = require('@aliceo2/web-ui');
-const { Kafka, logLevel } = require('kafkajs');
+const { Kafka } = require('kafkajs');
 const { KafkaConfig } = require('./config/index.js');
 const { AliEcsSynchronizer } = require('./server/kafka/AliEcsSynchronizer.js');
 const { environmentService } = require('./server/services/environment/EnvironmentService.js');
@@ -30,6 +31,7 @@ const { CcdbSynchronizer } = require('./server/externalServicesSynchronization/c
 const { promises: fs } = require('fs');
 const { MonAlisaClient } = require('./server/externalServicesSynchronization/monalisa/MonAlisaClient.js');
 const https = require('https');
+const { InfologgerKafkaLogCreator } = require('./server/kafka/InfologgerKafkaLogCreator.js');
 
 /**
  * Bookkeeping Application
@@ -50,7 +52,10 @@ class BookkeepingApplication {
             const kafkaClient = new Kafka({
                 clientId: 'bookkeeping',
                 brokers: KafkaConfig.brokers,
-                logLevel: logLevel.NOTHING,
+                logCreator: InfologgerKafkaLogCreator,
+                retry: {
+                    retries: Infinity,
+                },
             });
 
             this.aliEcsSynchronizer = new AliEcsSynchronizer({

lib/server/kafka/AliEcsSynchronizer.js

@@ -122,15 +122,16 @@
      */
     start() {
         this._logger.infoMessage('Starting to consume AliECS messages');
-        this.ecsEnvironmentsConsumer.start()
-            .catch((error) => this._logger.errorMessage(`Error when starting ECS environment consumer: ${error.message}\n${error.trace}`));
+        this.ecsEnvironmentsConsumer.start().catch((error) => this._logger.errorMessage(
+            `Error when starting ECS environment consumer: ${error.message}\n${JSON.stringify(error)}`,
+        ));
 
         this.ecsRunConsumer.start()
-            .catch((error) => this._logger.errorMessage(`Error when starting ECS runs consumer: ${error.message}\n${error.trace}`));
+            .catch((error) => this._logger.errorMessage(`Error when starting ECS runs consumer: ${error.message}\n${JSON.stringify(error)}`));
 
         this.ecsTrgIntegratedServiceConsumer.start()
             .catch((error) => this._logger
-                .errorMessage(`Error when starting ECS trigger integrated service consumer: ${error.message}\n${error.trace}`));
+                .errorMessage(`Error when starting ECS trigger integrated service consumer: ${error.message}\n${JSON.stringify(error)}`));
     }
 }
 

lib/server/kafka/InfologgerKafkaLogCreator.js

@@ -0,0 +1,52 @@
+/**
+ * @license
+ * Copyright CERN and copyright holders of ALICE O2. This software is
+ * distributed under the terms of the GNU General Public License v3 (GPL
+ * Version 3), copied verbatim in the file "COPYING".
+ *
+ * See http://alice-o2.web.cern.ch/license for full licensing information.
+ *
+ * In applying this license CERN does not waive the privileges and immunities
+ * granted to it by virtue of its status as an Intergovernmental Organization
+ * or submit itself to any jurisdiction.
+ */
+
+const { LogManager } = require('@aliceo2/web-ui');
+const { logLevel } = require('kafkajs');
+
+/**
+ * Kafka log creator to send messages to infologger
+ *
+ * @return {function} log creator ready to be used for infologger
+ * @constructor
+ */
+exports.InfologgerKafkaLogCreator = () => {
+    const logger = LogManager.getLogger('KAFKA');
+    return ({ level, log }) => {
+        const { broker, message, stack } = log;
+        let logMessage = '';
+        if (broker) {
+            logMessage = `Broker ${broker} - `;
+        }
+        logMessage += message;
+        if (stack) {
+            logMessage += `\n${stack}`;
+        }
+
+        switch (level) {
+            case logLevel.ERROR:
+            case logLevel.NOTHING:
+                logger.errorMessage(logMessage);
+                break;
+            case logLevel.WARN:
+                logger.warnMessage(logMessage);
+                break;
+            case logLevel.INFO:
+                logger.infoMessage(logMessage);
+                break;
+            default:
+                logger.debugMessage(logMessage);
+                break;
+        }
+    };
+};

test/public/defaults.js

@@ -393,14 +393,18 @@ module.exports.getInnerText = getInnerText;
  * @param {Object} page Puppeteer page object.
  * @param {string} selector Css selector.
  * @param {string} innerText Text to search for.
+ * @param {object} [options] eventual options
+ * @param {number} [options.timeout] Optional timeout
+ * @param {puppeteer.FrameWaitForFunctionOptions} [options.polling] Optional polling methods, see
  * @return {Promise<void>} resolves once the text has been checked
  */
-module.exports.expectInnerText = async (page, selector, innerText) => {
+module.exports.expectInnerText = async (page, selector, innerText, options = {}) => {
     const elementHandle = await page.waitForSelector(selector);
+
     try {
         await page.waitForFunction(
             (selector, innerText) => document.querySelector(selector).innerText === innerText,
-            {},
+            options,
             selector,
             innerText,
         );

test/public/runs/runsPerDataPass.overview.test.js

@@ -482,12 +482,19 @@ module.exports = () => {
             // eslint-disable-next-line no-undef
             sessionService.get().access.push(role);
         }, BkpRoles.DPG_ASYNC_QC_ADMIN);
-        await setConfirmationDialogToBeAccepted(page);
         const popoverSelector = await getPopoverSelector(await page.waitForSelector('#actions-dropdown-button .popover-trigger'));
         // Press again actions dropdown to re-trigger render
         await pressElement(page, '#actions-dropdown-button .popover-trigger', true);
+        await setConfirmationDialogToBeAccepted(page);
         await pressElement(page, `${popoverSelector} button:nth-child(3)`, true);
+        await pressElement(page, '#actions-dropdown-button .popover-trigger', true);
         await waitForTableLength(page, 3);
-        await expectInnerText(page, '#row106-CPV-text', 'QC'); // Expect QC flag button to be there
+        // Processing of data might take a bit of time, but then expect QC flag button to be there
+        await expectInnerText(
+            page,
+            '#row106-CPV-text',
+            'QC',
+            { timeout: 10000, polling: 'mutation' },
+        );
     });
 };