Skip to content

MINOR: MetadataShell should handle ProducerIdsRecord #11603

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

Merged
hachikuji merged 7 commits into from
Feb 12, 2022
Merged

MINOR: MetadataShell should handle ProducerIdsRecord #11603

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
51ac735
MINOR: MetadataShell should handle ProducerIdsRecord
Dec 14, 2021
29d8c62
fix comments
Dec 23, 2021
f17af95
resolve comments
dengziming Feb 10, 2022
81e6a9b
optimize path
dengziming Feb 11, 2022
4d97386
better naming
dengziming Feb 12, 2022
e7c832c
rebase
dengziming Feb 12, 2022
2888923
fix checkstyle
dengziming Feb 12, 2022
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
10 changes: 10 additions & 0 deletions shell/src/main/java/org/apache/kafka/shell/MetadataNodeManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.kafka.common.metadata.PartitionChangeRecord;
import org.apache.kafka.common.metadata.PartitionRecord;
import org.apache.kafka.common.metadata.PartitionRecordJsonConverter;
import org.apache.kafka.common.metadata.ProducerIdsRecord;
import org.apache.kafka.common.metadata.RegisterBrokerRecord;
import org.apache.kafka.common.metadata.RemoveTopicRecord;
import org.apache.kafka.common.metadata.TopicRecord;
Expand Down Expand Up @@ -318,6 +319,15 @@ private void handleCommitImpl(MetadataRecordType type, ApiMessage message)
node.create(record.key()).setContents(record.value() + "");
break;
}
case PRODUCER_IDS_RECORD: {
ProducerIdsRecord record = (ProducerIdsRecord) message;
DirectoryNode producerIds = data.root.mkdirs("producerIds");
producerIds.create("lastBlockBrokerId").setContents(record.brokerId() + "");
producerIds.create("lastBlockBrokerEpoch").setContents(record.brokerEpoch() + "");

producerIds.create("nextBlockStartId").setContents(record.nextProducerId() + "");
break;
}
default:
throw new RuntimeException("Unhandled metadata record type");
}
Expand Down
41 changes: 40 additions & 1 deletion shell/src/test/java/org/apache/kafka/shell/MetadataNodeManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.kafka.common.metadata.PartitionChangeRecord;
import org.apache.kafka.common.metadata.PartitionRecord;
import org.apache.kafka.common.metadata.PartitionRecordJsonConverter;
import org.apache.kafka.common.metadata.ProducerIdsRecord;
import org.apache.kafka.common.metadata.RegisterBrokerRecord;
import org.apache.kafka.common.metadata.RemoveTopicRecord;
import org.apache.kafka.common.metadata.TopicRecord;
Expand All @@ -35,6 +36,7 @@
import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.Collections;

import static org.apache.kafka.metadata.LeaderConstants.NO_LEADER_CHANGE;
import static org.junit.jupiter.api.Assertions.assertEquals;
Expand Down Expand Up @@ -276,7 +278,7 @@ public void testClientQuotaRecord() {
"user", "kraft").children().containsKey("producer_byte_rate"));

record = new ClientQuotaRecord()
.setEntity(Arrays.asList(
.setEntity(Collections.singletonList(
new ClientQuotaRecord.EntityData()
.setEntityType("user")
.setEntityName(null)
Expand All @@ -290,4 +292,41 @@ record = new ClientQuotaRecord()
metadataNodeManager.getData().root().directory("client-quotas",
"user", "<default>").file("producer_byte_rate").contents());
}

@Test
public void testProducerIdsRecord() {
// generate a producerId record
ProducerIdsRecord record1 = new ProducerIdsRecord()
.setBrokerId(0)
.setBrokerEpoch(1)
.setNextProducerId(10000);
metadataNodeManager.handleMessage(record1);

assertEquals(
"0",
metadataNodeManager.getData().root().directory("producerIds").file("lastBlockBrokerId").contents());
assertEquals(
"1",
metadataNodeManager.getData().root().directory("producerIds").file("lastBlockBrokerEpoch").contents());
assertEquals(
10000 + "",
metadataNodeManager.getData().root().directory("producerIds").file("nextBlockStartId").contents());

// generate another producerId record
ProducerIdsRecord record2 = new ProducerIdsRecord()
.setBrokerId(1)
.setBrokerEpoch(2)
.setNextProducerId(11000);
metadataNodeManager.handleMessage(record2);

assertEquals(
"1",
metadataNodeManager.getData().root().directory("producerIds").file("lastBlockBrokerId").contents());
assertEquals(
"2",
metadataNodeManager.getData().root().directory("producerIds").file("lastBlockBrokerEpoch").contents());
assertEquals(
11000 + "",
metadataNodeManager.getData().root().directory("producerIds").file("nextBlockStartId").contents());
}
}