root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Remove busy waiting in testTopicExpiryInMetadata

This commit is contained in:
ChangYu Huang 2025-08-19 16:24:09 -04:00
parent 2b68e44a35
commit f91dc5a239
1 changed files with 17 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
View File

@ -1047,8 +1047,9 @@ public class KafkaProducerTest {
@Test @Test
public void testTopicExpiryInMetadata() throws InterruptedException { public void testTopicExpiryInMetadata() throws InterruptedException {
Map<String, Object> configs = new HashMap<>(); Map<String, Object> configs = new HashMap<>();
final String maxBlockMs = "300000";
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000"); configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMs);
long refreshBackoffMs = 500L; long refreshBackoffMs = 500L;
long refreshBackoffMaxMs = 5000L; long refreshBackoffMaxMs = 5000L;
long metadataExpireMs = 60000L; long metadataExpireMs = 60000L;
@ -1057,28 +1058,33 @@ public class KafkaProducerTest {
final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs, final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
new LogContext(), new ClusterResourceListeners(), time); new LogContext(), new ClusterResourceListeners(), time);
final String topic = "topic"; final String topic = "topic";
final String warmupTopic = "warmup-topic";
MockClient client = new MockClient(time, metadata);
client.setShouldUpdateWithCurrentMetadata(false);
client.advanceTimeDuringPoll(true);
try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(), try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) { new StringSerializer(), metadata, client, null, time)) {
Exchanger<Void> exchanger = new Exchanger<>(); Exchanger<Void> exchanger = new Exchanger<>();
Thread t = new Thread(() -> { Thread t = new Thread(() -> {
try { try {
MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(warmupTopic, 1));
client.updateMetadata(updateResponse);
exchanger.exchange(null); // 1 exchanger.exchange(null); // 1
while (!metadata.updateRequested()) client.awaitMetadataUpdateRequest(Long.parseLong(maxBlockMs));
Thread.sleep(100); updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1)); client.updateMetadata(updateResponse);
metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
exchanger.exchange(null); // 2 exchanger.exchange(null); // 2
time.sleep(120 * 1000L); time.sleep(120 * 1000L);
// Update the metadata again, but it should be expired at this point. // Update the metadata again, but it should be expired at this point.
updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1)); updateResponse = RequestTestUtils.metadataUpdateWith(1, singletonMap(warmupTopic, 1));
metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds()); client.updateMetadata(updateResponse);
exchanger.exchange(null); // 3 exchanger.exchange(null); // 3
while (!metadata.updateRequested()) assertTrue(client.awaitMetadataUpdateRequest(Long.parseLong(maxBlockMs)));
Thread.sleep(100); System.out.println("Received metadata update request");
time.sleep(30 * 1000L); time.sleep(Long.parseLong(maxBlockMs));
} catch (Exception e) { } catch (Exception e) {
throw new RuntimeException(e); throw new RuntimeException(e);
} }