Delete Records Api implemented #141
Conversation
|
🎉 All Contributor License Agreements have been signed. Ready to merge. |
src/admin.cc
Outdated
|
|
||
| if (!info[0]->IsArray()) { | ||
| return Nan::ThrowError( | ||
| "Must provide array containg 'TopicPartition' objects"); |
There was a problem hiding this comment.
TopicPartitionOffset objects, right?
src/admin.cc
Outdated
| } | ||
|
|
||
| int operation_timeout_ms = | ||
| GetParameter<int64_t>(options, "operation_timeout", 6000); |
There was a problem hiding this comment.
Should be 60000 (60s) - make this change everywhere
| if (!this._isConnected) { | ||
| throw new Error('Client is disconnected'); | ||
| } | ||
|
|
There was a problem hiding this comment.
if (!options) {
options = {};
}Add this.
I also noticed this is missing in the above ListConsumerGroupOffsets method, add that also.
src/admin.cc
Outdated
| v8::Local<v8::Object> options = Nan::New<v8::Object>(); | ||
| if (info.Length() > 2 && info[1]->IsObject()) { | ||
| options = info[1].As<v8::Object>(); | ||
| } |
There was a problem hiding this comment.
No need for this. Just do
v8::Local<v8::Object> options = info[1].As<v8::Object>();
Same for ListConsumerGroupOffsets.
This will work with that change in admin.js which sets options to {} if it's unset.
lib/kafkajs/_admin.js
Outdated
| const topicPartitionOffsets = partitions.map(({ partition, offset }) => ({ | ||
| topic, | ||
| partition, | ||
| offset, |
There was a problem hiding this comment.
Offset is a string as per the type definition, you need to convert it.
So this needs to be in a loop, for example
for (const partition of partitions) {
if (partition.offset === null || partition.offset === undefined) // throw error
const offset = +partition.offset;
if (isNaN(offset)) // throw error
// add {topic, partition.partition, offset} to topicPartitionOffsets
}
| 'operation-timeout': { | ||
| type: 'string', | ||
| short: 'o', | ||
| default: '6000', |
| console.error("Partition and offset should be numbers and provided in pairs."); | ||
| process.exit(1); | ||
| } | ||
| partitions.push({ partition, offset }); |
There was a problem hiding this comment.
offset needs to actually be a string to comply with the API signature
| await producer.send({ topic: topicName, messages: messages }); | ||
|
|
||
| await expect( | ||
| admin.deleteTopicRecords({ topic: topicName, partitions: [{ partition: 0, offset: 4 }], timeout: 0 }) |
There was a problem hiding this comment.
Here and in all the other tests, the offset must be a string.
| partitions: [{ partition: 0, offset: 5 }], | ||
| }); | ||
|
|
||
| expect(records).toEqual([ |
There was a problem hiding this comment.
Remember to update the test after making error an optional field (omitted on success)
MIGRATION.md
Outdated
| * The `fetchOffsets` method is supported with additional `timeout` and | ||
| `requireStableOffsets` option but `resolveOffsets` option is not yet supported. | ||
| * The `deleteTopicRecords` method is supported with additional `timeout` | ||
| and `operationTiemout` option. |
There was a problem hiding this comment.
| and `operationTiemout` option. | |
| and `operationTimeout` options. |
src/admin.cc
Outdated
| v8::Local<v8::Array> delete_records_list = info[0].As<v8::Array>(); | ||
|
|
||
| if (delete_records_list->Length() == 0) { | ||
| return Nan::ThrowError("Must provide at least one TopicPartition"); |
There was a problem hiding this comment.
| return Nan::ThrowError("Must provide at least one TopicPartition"); | |
| return Nan::ThrowError("Must provide at least one TopicPartitionOffset"); |
| # confluent-kafka-javascript v0.5.0 | ||
|
|
||
| v0.4.0 is a limited availability feature release. It is supported for all usage. | ||
|
|
||
| ## Enhancements | ||
|
|
||
| 1. Add support for an Admin API to delete records.(#141). | ||
|
|
There was a problem hiding this comment.
| # confluent-kafka-javascript v0.5.0 | |
| v0.4.0 is a limited availability feature release. It is supported for all usage. | |
| ## Enhancements | |
| 1. Add support for an Admin API to delete records.(#141). | |
| # confluent-kafka-javascript v0.5.0 | |
| v0.5.0 is a limited availability feature release. It is supported for all usage. | |
| ## Enhancements | |
| 1. Add support for an Admin API to delete records.(#141). | |
Implemented delete records api in name of deleteTopicRecords and wrote tests and example for it.