chia7712 commented on code in PR #22083:
URL: https://github.com/apache/kafka/pull/22083#discussion_r3612852598
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -1003,91 +1002,75 @@ public void testTopicRefreshInMetadata() throws
InterruptedException {
}
@Test
- public void testTopicNotExistingInMetadata() throws InterruptedException {
+ public void testTopicNotExistingInMetadata() {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
- long refreshBackoffMs = 500L;
- long refreshBackoffMaxMs = 5000L;
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "2000");
+ configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
+ long refreshBackoffMs = 50L;
+ long refreshBackoffMaxMs = 500L;
long metadataExpireMs = 60000L;
long metadataIdleMs = 60000L;
- final Time time = new MockTime();
+ final Time time = Time.SYSTEM;
final ProducerMetadata metadata = new
ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs,
metadataIdleMs,
- new LogContext(), new ClusterResourceListeners(), time);
+ new LogContext(), new ClusterResourceListeners());
final String topic = "topic";
+ MockClient client = new MockClient(time, metadata);
+ // Seed initial metadata, then update with the topic marked as
UNKNOWN_TOPIC_OR_PARTITION
+ client.updateMetadata(RequestTestUtils.metadataUpdateWith(1,
Map.of()));
+ MetadataResponse errorResponse =
RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
+ singletonMap(topic, Errors.UNKNOWN_TOPIC_OR_PARTITION),
emptyMap());
+ client.prepareMetadataUpdate(errorResponse);
try (KafkaProducer<String, String> producer = kafkaProducer(configs,
new StringSerializer(),
- new StringSerializer(), metadata, new MockClient(time,
metadata), null, time)) {
-
- Exchanger<Void> exchanger = new Exchanger<>();
+ new StringSerializer(), metadata, client, null, time)) {
- Thread t = new Thread(() -> {
- try {
- // Update the metadata with non-existing topic.
- MetadataResponse updateResponse =
RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
- singletonMap(topic,
Errors.UNKNOWN_TOPIC_OR_PARTITION), emptyMap());
- metadata.updateWithCurrentRequestVersion(updateResponse,
false, time.milliseconds());
- exchanger.exchange(null);
- while (!metadata.updateRequested())
- Thread.sleep(100);
- time.sleep(30 * 1000L);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- t.start();
- exchanger.exchange(null);
+ // partitionsFor should time out via real wait() because the topic
has an error
+ // and the Sender keeps replaying the same error metadata.
Throwable throwable = assertThrows(TimeoutException.class, () ->
producer.partitionsFor(topic));
assertInstanceOf(UnknownTopicOrPartitionException.class,
throwable.getCause());
- t.join();
}
}
@Test
public void testTopicExpiryInMetadata() throws InterruptedException {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
- long refreshBackoffMs = 500L;
- long refreshBackoffMaxMs = 5000L;
- long metadataExpireMs = 60000L;
- long metadataIdleMs = 60000L;
- final Time time = new MockTime();
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "2000");
+ configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");
+ long refreshBackoffMs = 50L;
+ long refreshBackoffMaxMs = 500L;
+ long metadataExpireMs = 1000L;
+ long metadataIdleMs = 1000L;
+ final Time time = Time.SYSTEM;
final ProducerMetadata metadata = new
ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs,
metadataIdleMs,
- new LogContext(), new ClusterResourceListeners(), time);
+ new LogContext(), new ClusterResourceListeners());
final String topic = "topic";
+ MockClient client = new MockClient(time, metadata);
+ // Seed initial metadata without the topic
+ client.updateMetadata(RequestTestUtils.metadataUpdateWith(1,
Map.of()));
+ // Queue a metadata response with the topic for the first
partitionsFor call
+ client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWith(1,
Map.of(topic, 1)));
+ // Queue a follow-up without the topic so that after consumption,
updateWithCurrentMetadata
+ // replays the empty response
+ client.prepareMetadataUpdate(RequestTestUtils.metadataUpdateWith(1,
Map.of()));
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<>();
+ // First call should succeed — the queued metadata response
includes the topic
+ assertNotNull(producer.partitionsFor(topic));
- Thread t = new Thread(() -> {
- try {
- exchanger.exchange(null); // 1
- while (!metadata.updateRequested())
- Thread.sleep(100);
- MetadataResponse updateResponse =
RequestTestUtils.metadataUpdateWith(1, singletonMap(topic, 1));
- metadata.updateWithCurrentRequestVersion(updateResponse,
false, time.milliseconds());
- exchanger.exchange(null); // 2
- time.sleep(120 * 1000L);
+ // Wait for topic metadata to expire (metadataIdleMs = 1000ms)
+ Thread.sleep(1500);
Review Comment:
This makes the test slightly slower, but I think our CI can handle the extra
delay
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java:
##########
@@ -138,14 +136,19 @@ public synchronized boolean retainTopic(String topic,
boolean isInternal, long n
/**
* Wait for metadata update until the current version is larger than the
last version we know of
*/
- public synchronized void awaitUpdate(final int lastVersion, final long
timeoutMs) throws InterruptedException {
- long currentTimeMs = time.milliseconds();
- long deadlineMs = currentTimeMs + timeoutMs < 0 ? Long.MAX_VALUE :
currentTimeMs + timeoutMs;
- time.waitObject(this, () -> {
Review Comment:
`waitObject` is now dead code, so we can remove the related tests and add a
default implementation to Time, right?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
