Remove kafkaMaxRequestSize from oplog populator

Will be set globally in kafka-connect (via ZKOP), and allows custom values to be (manually) set on each connector if ever needed: as any "extra" config will be preserved by oplog populator. Issue: BB-436
scality · Sep 11, 2023 · f0a23f5 · f0a23f5
1 parent 6482189
commit f0a23f5
Show file tree

Hide file tree

Showing 4 changed files with 1 addition and 24 deletions.
diff --git a/extensions/oplogPopulator/OplogPopulator.js b/extensions/oplogPopulator/OplogPopulator.js
@@ -13,9 +13,7 @@ const { OplogPopulatorConfigJoiSchema } = require('./OplogPopulatorConfigValidat
 const { mongoJoi } = require('../../lib/config/configItems.joi');
 
 const paramsJoi = joi.object({
-    config: OplogPopulatorConfigJoiSchema.keys({
-        maxRequestSize: joi.number().required(),
-    }).required(),
+    config: OplogPopulatorConfigJoiSchema.required(),
     mongoConfig: mongoJoi.required(),
     activeExtensions: joi.array().required(),
     logger: joi.object().required(),
@@ -34,7 +32,6 @@ class OplogPopulator {
      * @constructor
      * @param {Object} params - constructor params
      * @param {Object} params.config - oplog populator config
-     * @param {Object} params.config.maxRequestSize - kafka producer's max request size
      * @param {Object} params.mongoConfig - mongo connection config
      * @param {Object} params.mongoConfig.authCredentials - mongo auth credentials
      * @param {Object} params.mongoConfig.replicaSetHosts - mongo replication hosts
@@ -264,7 +261,6 @@ class OplogPopulator {
                heartbeatIntervalMs: this._config.heartbeatIntervalMs,
                kafkaConnectHost: this._config.kafkaConnectHost,
                kafkaConnectPort: this._config.kafkaConnectPort,
-               kafkaMaxRequestSize: this._maxRequestSize,
                metricsHandler: this._metricsHandler,
                logger: this._logger,
             });

diff --git a/extensions/oplogPopulator/modules/ConnectorsManager.js b/extensions/oplogPopulator/modules/ConnectorsManager.js
@@ -20,7 +20,6 @@ const paramsJoi = joi.object({
     heartbeatIntervalMs: joi.number().required(),
     kafkaConnectHost: joi.string().required(),
     kafkaConnectPort: joi.number().required(),
-    kafkaMaxRequestSize: joi.number().required(),
     metricsHandler: joi.object()
         .instance(OplogPopulatorMetrics).required(),
     logger: joi.object().required(),
@@ -62,7 +61,6 @@ class ConnectorsManager {
             kafkaConnectPort: this._kafkaConnectPort,
             logger: this._logger,
         });
-        this._kafkaMaxRequestSize = params.kafkaMaxRequestSize;
         this._metricsHandler = params.metricsHandler;
         this._database = params.database;
         this._mongoUrl = params.mongoUrl;
@@ -89,7 +87,6 @@ class ConnectorsManager {
             // hearbeat prevents having an outdated resume token in the connectors
             // by constantly updating the offset to the last object in the oplog
             'heartbeat.interval.ms': this._heartbeatIntervalMs,
-            'producer.override.max.request.size': this._kafkaMaxRequestSize,
         };
         return {
             ...constants.defaultConnectorConfig,

diff --git a/tests/unit/oplogPopulator/ConnectorsManager.js b/tests/unit/oplogPopulator/ConnectorsManager.js
@@ -56,7 +56,6 @@ const connectorConfig = {
         }],
     }),
     'heartbeat.interval.ms': 10000,
-    'producer.override.max.request.size': 5000020,
 };
 
 describe('ConnectorsManager', () => {
@@ -92,7 +91,6 @@ describe('ConnectorsManager', () => {
             heartbeatIntervalMs: 10000,
             kafkaConnectHost: '127.0.0.1',
             kafkaConnectPort: 8083,
-            kafkaMaxRequestSize: 5000020,
             metricsHandler: new OplogPopulatorMetrics(logger),
             logger,
         });

diff --git a/tests/unit/oplogPopulator/oplogPopulator.js b/tests/unit/oplogPopulator/oplogPopulator.js
@@ -18,7 +18,6 @@ const oplogPopulatorConfig = {
     kafkaConnectPort: 8083,
     numberOfConnectors: 1,
     probeServer: { port: 8552 },
-    maxRequestSize: 5000020,
 };
 
 const mongoConfig = {
@@ -93,19 +92,6 @@ describe('OplogPopulator', () => {
                 activeExtensions,
                 logger,
             }));
-            assert.throws(() => new OplogPopulator({
-                config: {
-                    topic: 'oplog',
-                    kafkaConnectHost: '127.0.0.1',
-                    kafkaConnectPort: 8083,
-                    numberOfConnectors: 1,
-                    probeServer: { port: 8552 },
-                },
-                mongoConfig,
-                activeExtensions,
-                enableMetrics: false,
-                logger,
-            }));
             const op = new OplogPopulator({
                 config: oplogPopulatorConfig,
                 mongoConfig,