showuon commented on code in PR #15917:
URL: https://github.com/apache/kafka/pull/15917#discussion_r1598332128
##########
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataManagerTestUtils.java:
##########
@@ -76,10 +77,17 @@ public Builder
overrideRemoteLogMetadataManagerProps(Map<String, Object> overrid
return this;
}
+ public Builder topicIdPartitions(Set<TopicIdPartition>
topicIdPartitions) {
+ this.topicIdPartitions = Objects.requireNonNull(topicIdPartitions);
+ return this;
+ }
+
public TopicBasedRemoteLogMetadataManager build() {
+ Objects.requireNonNull(bootstrapServers);
String logDir =
TestUtils.tempDirectory("rlmm_segs_").getAbsolutePath();
TopicBasedRemoteLogMetadataManager
topicBasedRemoteLogMetadataManager = new
TopicBasedRemoteLogMetadataManager(startConsumerThread);
+ System.out.println("[CHIA] " + bootstrapServers);
Review Comment:
debug log
##########
storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java:
##########
@@ -68,52 +48,13 @@ public void initialize(Set<TopicIdPartition>
topicIdPartitions,
public void initializeRemoteLogMetadataManager(Set<TopicIdPartition>
topicIdPartitions,
boolean startConsumerThread,
RemoteLogMetadataTopicPartitioner remoteLogMetadataTopicPartitioner) {
- String logDir =
TestUtils.tempDirectory("rlmm_segs_").getAbsolutePath();
- topicBasedRemoteLogMetadataManager = new
TopicBasedRemoteLogMetadataManager(startConsumerThread) {
- @Override
- public void onPartitionLeadershipChanges(Set<TopicIdPartition>
leaderPartitions,
- Set<TopicIdPartition>
followerPartitions) {
- Set<TopicIdPartition> allReplicas = new
HashSet<>(leaderPartitions);
- allReplicas.addAll(followerPartitions);
- // Make sure the topic partition dirs exist as the topics
might not have been created on this broker.
- for (TopicIdPartition topicIdPartition : allReplicas) {
- // Create partition directory in the log directory created
by topicBasedRemoteLogMetadataManager.
- File partitionDir = new File(new File(config().logDir()),
topicIdPartition.topicPartition().topic() + "-" +
topicIdPartition.topicPartition().partition());
- partitionDir.mkdirs();
- if (!partitionDir.exists()) {
- throw new KafkaException("Partition directory:[" +
partitionDir + "] could not be created successfully.");
- }
- }
-
- super.onPartitionLeadershipChanges(leaderPartitions,
followerPartitions);
- }
- };
-
- // Initialize TopicBasedRemoteLogMetadataManager.
- Map<String, Object> configs = new HashMap<>();
- configs.put(REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers(listenerName()));
- configs.put(BROKER_ID, 0);
- configs.put(LOG_DIR, logDir);
- configs.put(REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP,
METADATA_TOPIC_PARTITIONS_COUNT);
- configs.put(REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR_PROP,
METADATA_TOPIC_REPLICATION_FACTOR);
- configs.put(REMOTE_LOG_METADATA_TOPIC_RETENTION_MS_PROP,
METADATA_TOPIC_RETENTION_MS);
-
- log.debug("TopicBasedRemoteLogMetadataManager configs before adding
overridden properties: {}", configs);
- // Add override properties.
- configs.putAll(overrideRemoteLogMetadataManagerProps());
- log.debug("TopicBasedRemoteLogMetadataManager configs after adding
overridden properties: {}", configs);
-
- topicBasedRemoteLogMetadataManager.configure(configs);
- if (remoteLogMetadataTopicPartitioner != null) {
-
topicBasedRemoteLogMetadataManager.setRlmTopicPartitioner(remoteLogMetadataTopicPartitioner);
- }
- try {
-
RemoteLogMetadataManagerTestUtils.waitUntilInitialized(this.remoteLogMetadataManager(),
60_000);
- } catch (InterruptedException e) {
- throw new KafkaException(e);
- }
+ topicBasedRemoteLogMetadataManager =
RemoteLogMetadataManagerTestUtils.builder()
+ .topicIdPartitions(topicIdPartitions)
+ .bootstrapServers(bootstrapServers(listenerName()))
+ .startConsumerThread(startConsumerThread)
+
.remoteLogMetadataTopicPartitioner(remoteLogMetadataTopicPartitioner)
+ .build();
Review Comment:
Thanks for the refactor!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
