Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Check if GC config exists before creating gc producer in replication #2555

Merged
merged 2 commits into from
Oct 23, 2024
Merged

Check if GC config exists before creating gc producer in replication #2555

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
daf94e2
Check if GC config exists before creating gc producer in replication
KillianG Oct 15, 2024
123f50d
Add functionnal tests for RSP gc producer
KillianG Oct 22, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 9 additions & 5 deletions extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,14 +204,16 @@
* in PEM format
* @param {Object} mConfig - metrics config
* @param {String} mConfig.topic - metrics config kafka topic
* @param {Object} gcConfig - config for garbage collector
*/
constructor(kafkaConfig, sourceConfig, repConfig,
internalHttpsConfig, mConfig) {
internalHttpsConfig, mConfig, gcConfig) {
this.kafkaConfig = kafkaConfig;
this.sourceConfig = sourceConfig;
this.repConfig = repConfig;
this.internalHttpsConfig = internalHttpsConfig;
this.mConfig = mConfig;
this.gcConfig = gcConfig;
this._consumer = null;
this._gcProducer = null;
this._mProducer = null;
Expand Down Expand Up @@ -370,8 +372,12 @@
});
},
done => {
this._gcProducer = new GarbageCollectorProducer();
this._gcProducer.setupProducer(done);
if (this.gcConfig) {
this._gcProducer = new GarbageCollectorProducer();
this._gcProducer.setupProducer(done);
} else {
done();

Check warning on line 379 in extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js

View check run for this annotation

Codecov / codecov/patch/Backbeat

extensions/replication/replicationStatusProcessor/ReplicationStatusProcessor.js#L379 

Added line #L379 was not covered by tests
}
KillianG marked this conversation as resolved.
Show resolved Hide resolved
},
done => {
this._mProducer = new MetricsProducer(this.kafkaConfig,
Expand Down Expand Up @@ -458,8 +464,6 @@
processKafkaEntry(kafkaEntry, done) {
const sourceEntry = QueueEntry.createFromKafkaEntry(kafkaEntry);
if (sourceEntry.error) {
this.logger.error('error processing source entry',
{ error: sourceEntry.error });
return process.nextTick(() => done(errors.InternalError));
KillianG marked this conversation as resolved.
Show resolved Hide resolved
}
let task;
Expand Down
3 changes: 2 additions & 1 deletion extensions/replication/replicationStatusProcessor/task.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,12 @@
const sourceConfig = repConfig.source;
const internalHttpsConfig = config.internalHttps;
const mConfig = config.metrics;
const gcConfig = config.extensions.gc;

Check warning on line 16 in extensions/replication/replicationStatusProcessor/task.js

View check run for this annotation

Codecov / codecov/patch/Backbeat

extensions/replication/replicationStatusProcessor/task.js#L16 

Added line #L16 was not covered by tests

const { initManagement } = require('../../../lib/management/index');

const replicationStatusProcessor = new ReplicationStatusProcessor(
kafkaConfig, sourceConfig, repConfig, internalHttpsConfig, mConfig);
kafkaConfig, sourceConfig, repConfig, internalHttpsConfig, mConfig, gcConfig);

werelogs.configure({ level: config.log.logLevel,
dump: config.log.dumpLevel });
Expand Down
7 changes: 4 additions & 3 deletions tests/functional/replication/queueProcessor.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@ const replicationConstants = require('../../../extensions/replication/constants'
const ReplicationAPI =
require('../../../extensions/replication/ReplicationAPI');
const TestConfigurator = require('../../utils/TestConfigurator');
const ReplicationStatusProcessor = require('../../../extensions/replication' +
'/replicationStatusProcessor/ReplicationStatusProcessor');

/* eslint-disable max-len */

Expand Down Expand Up @@ -760,8 +762,6 @@ describe('queue processor functional tests with mocking', () => {
let copyLocationResultsConsumer;

const QueueProcessor = require('../../../extensions/replication/queueProcessor/QueueProcessor');
const ReplicationStatusProcessor = require('../../../extensions/replication' +
'/replicationStatusProcessor/ReplicationStatusProcessor');

before(function before(done) {
this.timeout(60000);
Expand Down Expand Up @@ -843,7 +843,8 @@ describe('queue processor functional tests with mocking', () => {
objectSizeMetrics: [100, 1000],
},
{},
{ topic: 'metrics-test-topic' });
{ topic: 'metrics-test-topic' },
{ topic: 'backbeat-gc' });
replicationStatusProcessor.start({ bootstrap: true }, done);
},
done => {
Expand Down
Loading