We have a huge enterprise hospital information management system application which is built in Java 8 and mysql as database
That application is called medics.
In medics we can do many types of transactions like bills, receipts, advances, refunds, generate lab reports, goods receipt notes, etc…
There are many of our sub products which wanted this information in real time.
So we implemented kafka and producing the events whenever any transaction was done in medics.
This was working but some times due to some issues kafka is getting stopped and in the medics application we coudlnt able to do any transactions and it is becoming a panic situation because no transactions are happening and patients are waiting in the hospitals. so we have to suggest them to do manual transactions by delinking kafka for some time till it got up.
and also when the kafka started back, there is no way we can get those transactions back and produce them.

So what we are thinking of doing is whenever the transaction is done
we will store the complete message in our mysql table and thenin another thread kafka will take it from there and publish it and then update the status for that message as delivered or some thing

We decided to do this program in node js like below


const mysql = require('mysql2/promise'); const Kafka = require('node-rdkafka'); const { promisify } = require('util'); const winston = require('winston'); const DailyRotateFile = require('winston-daily-rotate-file'); const express = require('express'); // Create a logger with Winston const logger = winston.createLogger({ transports: [ new DailyRotateFile({ level: 'info', filename: 'logs/application-%DATE%.log', datePattern: 'YYYY-MM-DD', zippedArchive: true, maxSize: '20m', maxFiles: '14d', // Keep logs for 14 days json: true, }), ], }); // MySQL configuration const mysqlConfig = { host: process.env.MYSQL_HOST || 'localhost', user: process.env.MYSQL_USER || 'root', password: process.env.MYSQL_PASSWORD || 'password', database: process.env.MYSQL_DATABASE || 'your-database-name', table: process.env.MYSQL_TABLE || 'notification_queue', }; // Kafka configuration const kafkaConfig = { 'metadata.broker.list': process.env.KAFKA_BROKERS || 'localhost:9092', 'group.id': process.env.KAFKA_GROUP || 'kafka-group', 'enable.auto.commit': false, }; const consumer = new Kafka.KafkaConsumer(kafkaConfig); // Flag to prevent concurrent polling let isPolling = false; // Lock to ensure the application runs only once let isAppRunning = false; // Set to store processed message IDs and prevent duplicates const processedMessageIds = new Set(); async function connectToDatabase() { const connection = await mysql.createConnection(mysqlConfig); return connection; } function createKafkaProducer() { const producer = new Kafka.Producer(kafkaConfig); producer.connect(); return producer; } async function publishMessage(producer, topic, payload) { const produceAsync = promisify(producer.produce.bind(producer)); try { await produceAsync(topic, null, Buffer.from(payload)); logger.info('Message published to Kafka successfully.', { payload }); } catch (error) { logger.error('Error publishing message to Kafka:', { error, payload }); // Optionally, you might want to implement a retry mechanism here } } async function publishMessageWithRetry(producer, topic, payload, retryCount = 3) { const messageId = /* Generate a unique identifier for the message */; if (processedMessageIds.has(messageId)) { logger.warn(`Duplicate message detected. Skipping: ${payload}`); return; } for (let attempt = 1; attempt <= retryCount; attempt++) { try { await publishMessage(producer, topic, payload); processedMessageIds.add(messageId); break; // Break the loop if successful } catch (error) { logger.warn(`Error publishing message to Kafka (Attempt ${attempt}):`, { error, payload }); if (attempt === retryCount) { logger.error(`Max retry attempts reached. Skipping message: ${payload}`); // Log or handle the fact that the message could not be sent after multiple retries logMessageSendFailure(payload); } else { // Optionally, you might want to add a delay between retry attempts await new Promise((resolve) => setTimeout(resolve, 1000)); } } } } async function logMessageSendFailure(payload) { // Add your custom logic for logging or handling the failure logger.error(`Message could not be sent after multiple retries: ${payload}`); // You might also want to notify administrators or take additional actions here } async function fetchAndPublishRecords() { if (isPolling) { logger.warn('Previous polling is still in progress. Skipping current poll.'); return; } if (isAppRunning) { logger.warn('The application is already running. Skipping current execution.'); return; } isAppRunning = true; const connection = await connectToDatabase(); const producer = createKafkaProducer(); try { // Fetch a batch of records from the notification_queue table const [rows] = await connection.execute(`SELECT * FROM ${mysqlConfig.table} WHERE STATUS = 'pending' LIMIT 100`); if (rows.length > 0) { // Create an array to store produced messages const producedMessages = []; // Produce messages to Kafka with retry await Promise.all( rows.map(async (row) => { const payload = JSON.stringify(row); try { await publishMessageWithRetry(producer, 'your-kafka-topic', payload); // Update status, delivered_at, and error_message fields in the notification_queue table await connection.execute( `UPDATE ${mysqlConfig.table} SET STATUS = 'delivered', delivered_at = CURRENT_TIMESTAMP(), error_message = NULL WHERE id = ?`, [row.id] ); producedMessages.push(row.id); } catch (error) { // If an error occurs during publishing, update the error_message field in the notification_queue table await connection.execute( `UPDATE ${mysqlConfig.table} SET STATUS = 'failed', delivered_at = CURRENT_TIMESTAMP(), error_message = ? WHERE id = ?`, [error.message, row.id] ); logger.warn(`Error publishing message to Kafka: ${error.message}`, { payload }); } }) ); // Await producer flush for proper control of the message send process await new Promise((resolve) => { producer.flush(5000, (err) => { if (err) { logger.error('Error flushing Kafka producer:', { error: err }); // Optionally, you might want to handle the error or retry producing messages } else { logger.info('Records published to Kafka successfully.'); } resolve(); }); }); // Optionally, update status to 'delivered' in the notification_queue table for the produced messages if (producedMessages.length > 0) { const ids = producedMessages.join(','); await connection.execute(`UPDATE ${mysqlConfig.table} SET STATUS = 'delivered' WHERE id IN (${ids})`); } } } catch (error) { logger.error('Error fetching and publishing records:', { error }); // Update error_message field in the notification_queue table for pending records await connection.execute( `UPDATE ${mysqlConfig.table} SET STATUS = 'failed', error_message = ? WHERE STATUS = 'pending'`, [error.message] ); } finally { await connection.end(); isAppRunning = false; isPolling = false; } } // Poll for records and publish every 1 minute (configurable) setInterval(fetchAndPublishRecords, process.env.POLL_INTERVAL || 60000); // Create a health check endpoint const app = express(); const port = process.env.PORT || 3000; app.get('/health', (req, res) => { const healthStatus = { status: 'ok', isPolling, isAppRunning, processedMessageCount: processedMessageIds.size, // Add any additional health-related information here }; res.json(healthStatus); }); app.listen(port, () => { logger.info(`Health check server listening at http://localhost:${port}/health`); });

our application has to handle thousands of transcations per second,
so this node js thread should pick them and produce them in almost real time.
so is this single file enough to handle such huge data or should we need to re structure this ?