You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
334 lines
9.0 KiB
334 lines
9.0 KiB
/*
|
|
* node-rdkafka - Node.js wrapper for RdKafka C/C++ library
|
|
*
|
|
* Copyright (c) 2016 Blizzard Entertainment
|
|
*
|
|
* This software may be modified and distributed under the terms
|
|
* of the MIT license. See the LICENSE.txt file for details.
|
|
*/
|
|
|
|
var Kafka = require('../');
|
|
|
|
var kafkaBrokerList = process.env.KAFKA_HOST || 'localhost:9092';
|
|
|
|
describe('Transactional Producer', function () {
|
|
this.timeout(5000);
|
|
var TRANSACTIONS_TIMEOUT_MS = 30000;
|
|
var r = Date.now() + '_' + Math.round(Math.random() * 1000);
|
|
var topicIn = 'transaction_input_' + r;
|
|
var topicOut = 'transaction_output_' + r;
|
|
|
|
var producerTras;
|
|
var consumerTrans;
|
|
|
|
before(function (done) {
|
|
/*
|
|
prepare:
|
|
transactional consumer (read from input topic)
|
|
transactional producer (write to output topic)
|
|
write 3 messages to input topic: A, B, C
|
|
A will be skipped, B will be committed, C will be aborted
|
|
*/
|
|
var connecting = 3;
|
|
var producerInput;
|
|
function connectedCb(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
connecting--;
|
|
if (connecting === 0) {
|
|
producerInput.produce(topicIn, -1, Buffer.from('A'));
|
|
producerInput.produce(topicIn, -1, Buffer.from('B'));
|
|
producerInput.produce(topicIn, -1, Buffer.from('C'));
|
|
producerInput.disconnect(function (err) {
|
|
consumerTrans.subscribe([topicIn]);
|
|
done(err);
|
|
})
|
|
}
|
|
}
|
|
producerInput = Kafka.Producer({
|
|
'client.id': 'kafka-test',
|
|
'metadata.broker.list': kafkaBrokerList,
|
|
'enable.idempotence': true
|
|
});
|
|
producerInput.setPollInterval(100);
|
|
producerInput.connect({}, connectedCb);
|
|
|
|
producerTras = new Kafka.Producer({
|
|
'client.id': 'kafka-test',
|
|
'metadata.broker.list': kafkaBrokerList,
|
|
'dr_cb': true,
|
|
'debug': 'all',
|
|
'transactional.id': 'noderdkafka_transactions_send_offset',
|
|
'enable.idempotence': true
|
|
});
|
|
producerTras.setPollInterval(100);
|
|
producerTras.connect({}, connectedCb);
|
|
|
|
consumerTrans = new Kafka.KafkaConsumer({
|
|
'metadata.broker.list': kafkaBrokerList,
|
|
'group.id': 'gropu_transaction_consumer',
|
|
'enable.auto.commit': false
|
|
}, {
|
|
'auto.offset.reset': 'earliest',
|
|
});
|
|
consumerTrans.connect({}, connectedCb);
|
|
});
|
|
|
|
after(function (done) {
|
|
let connected = 2;
|
|
function execDisconnect(client) {
|
|
if (!client.isConnected) {
|
|
connected--;
|
|
if (connected === 0) {
|
|
done();
|
|
}
|
|
} else {
|
|
client.disconnect(function() {
|
|
connected--;
|
|
if (connected === 0) {
|
|
done();
|
|
}
|
|
});
|
|
}
|
|
}
|
|
execDisconnect(producerTras);
|
|
execDisconnect(consumerTrans);
|
|
});
|
|
|
|
it('should init transactions', function(done) {
|
|
producerTras.initTransactions(TRANSACTIONS_TIMEOUT_MS, function (err) {
|
|
done(err);
|
|
});
|
|
});
|
|
|
|
it('should complete transaction', function(done) {
|
|
function readMessage() {
|
|
consumerTrans.consume(1, function(err, m) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
if (m.length === 0) {
|
|
readMessage();
|
|
} else {
|
|
var v = m[0].value.toString();
|
|
if (v === 'A') { // skip first message
|
|
readMessage();
|
|
return;
|
|
}
|
|
if (v !== 'B') {
|
|
done('Expected B');
|
|
return;
|
|
}
|
|
producerTras.beginTransaction(function (err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
producerTras.produce(topicOut, -1, Buffer.from(v));
|
|
var position = consumerTrans.position();
|
|
producerTras.sendOffsetsToTransaction(position, consumerTrans, function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
producerTras.commitTransaction(function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
consumerTrans.committed(5000, function(err, tpo) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
if (JSON.stringify(position) !== JSON.stringify(tpo)) {
|
|
done('Committed mismatch');
|
|
return;
|
|
}
|
|
done();
|
|
});
|
|
});
|
|
});
|
|
});
|
|
}
|
|
});
|
|
}
|
|
readMessage();
|
|
});
|
|
|
|
describe('abort transaction', function() {
|
|
var lastConsumerTransPosition;
|
|
before(function(done) {
|
|
function readMessage() {
|
|
consumerTrans.consume(1, function(err, m) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
if (m.length === 0) {
|
|
readMessage();
|
|
} else {
|
|
var v = m[0].value.toString();
|
|
if (v !== 'C') {
|
|
done('Expected C');
|
|
return;
|
|
}
|
|
producerTras.beginTransaction(function (err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
producerTras.produce(topicOut, -1, Buffer.from(v));
|
|
lastConsumerTransPosition = consumerTrans.position();
|
|
producerTras.sendOffsetsToTransaction(lastConsumerTransPosition, consumerTrans, function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
done();
|
|
});
|
|
});
|
|
}
|
|
});
|
|
}
|
|
readMessage();
|
|
});
|
|
|
|
it ('should consume committed and uncommitted for read_uncommitted', function(done) {
|
|
var allMsgs = [];
|
|
var consumer = new Kafka.KafkaConsumer({
|
|
'metadata.broker.list': kafkaBrokerList,
|
|
'group.id': 'group_read_uncommitted',
|
|
'enable.auto.commit': false,
|
|
'isolation.level': 'read_uncommitted'
|
|
}, {
|
|
'auto.offset.reset': 'earliest',
|
|
});
|
|
consumer.connect({}, function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
consumer.subscribe([topicOut]);
|
|
consumer.consume();
|
|
});
|
|
consumer.on('data', function(msg) {
|
|
var v = msg.value.toString();
|
|
allMsgs.push(v);
|
|
// both B and C must be consumed
|
|
if (allMsgs.length === 2 && allMsgs[0] === 'B' && allMsgs[1] === 'C') {
|
|
consumer.disconnect(function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
done();
|
|
})
|
|
}
|
|
});
|
|
});
|
|
|
|
it ('should consume only committed for read_committed', function(done) {
|
|
var allMsgs = [];
|
|
var consumer = new Kafka.KafkaConsumer({
|
|
'metadata.broker.list': kafkaBrokerList,
|
|
'group.id': 'group_read_committed',
|
|
'enable.partition.eof': true,
|
|
'enable.auto.commit': false,
|
|
'isolation.level': 'read_committed'
|
|
}, {
|
|
'auto.offset.reset': 'earliest',
|
|
});
|
|
consumer.connect({}, function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
consumer.subscribe([topicOut]);
|
|
consumer.consume();
|
|
});
|
|
consumer.on('data', function(msg) {
|
|
var v = msg.value.toString();
|
|
allMsgs.push(v);
|
|
});
|
|
consumer.on('partition.eof', function(eof) {
|
|
if (allMsgs.length === 1 && allMsgs[0] === 'B') {
|
|
consumer.disconnect(function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
done();
|
|
})
|
|
} else {
|
|
done('Expected only B');
|
|
return;
|
|
}
|
|
});
|
|
});
|
|
|
|
it('should abort transaction', function(done) {
|
|
producerTras.abortTransaction(function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
consumerTrans.committed(5000, function(err, tpo) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
if (lastConsumerTransPosition[0].offset <= tpo[0].offset) {
|
|
done('Committed mismatch');
|
|
return;
|
|
}
|
|
done();
|
|
});
|
|
});
|
|
});
|
|
|
|
it('should consume only committed', function(done) {
|
|
var gotB = false;
|
|
var consumer = new Kafka.KafkaConsumer({
|
|
'metadata.broker.list': kafkaBrokerList,
|
|
'group.id': 'group_default',
|
|
'enable.partition.eof': true,
|
|
'enable.auto.commit': false,
|
|
}, {
|
|
'auto.offset.reset': 'earliest',
|
|
});
|
|
consumer.connect({}, function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
consumer.subscribe([topicOut]);
|
|
consumer.consume();
|
|
});
|
|
consumer.on('data', function(msg) {
|
|
var v = msg.value.toString();
|
|
if (v !== 'B') {
|
|
done('Expected B');
|
|
return;
|
|
}
|
|
gotB = true;
|
|
});
|
|
consumer.on('partition.eof', function(eof) {
|
|
if (!gotB) {
|
|
done('Expected B');
|
|
return;
|
|
}
|
|
consumer.disconnect(function(err) {
|
|
if (err) {
|
|
done(err);
|
|
return;
|
|
}
|
|
done();
|
|
});
|
|
});
|
|
});
|
|
});
|
|
});
|