This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 36a69b49c5b KAFKA-16937 Inline Time#waitObject to
ProducerMetadata#awaitUpdate (#22083)
36a69b49c5b is described below
commit 36a69b49c5b4bfdba775f54a9dc08bebde273081
Author: PoAn Yang <[email protected]>
AuthorDate: Mon Jul 27 12:19:44 2026 +0900
KAFKA-16937 Inline Time#waitObject to ProducerMetadata#awaitUpdate (#22083)
Inline `Time#waitObject` to `ProducerMetadata#awaitUpdate` and use
`Timer` in `ProducerMetadata#awaitUpdate`.
Reviewers: Chia-Ping Tsai <[email protected]>, Sean Quah
<[email protected]>
---
.../kafka/clients/producer/KafkaProducer.java | 7 +-
.../producer/internals/ProducerMetadata.java | 25 ++--
.../org/apache/kafka/common/utils/SystemTime.java | 20 ---
.../java/org/apache/kafka/common/utils/Time.java | 15 ++-
.../kafka/clients/producer/KafkaProducerTest.java | 142 +++++++++------------
.../producer/internals/ProducerMetadataTest.java | 12 +-
.../clients/producer/internals/SenderTest.java | 2 +-
.../producer/internals/TransactionManagerTest.java | 2 +-
.../apache/kafka/common/utils/MockTimeTest.java | 7 +-
.../apache/kafka/common/utils/SystemTimeTest.java | 25 ----
.../org/apache/kafka/common/utils/TimeTest.java | 15 +--
.../org/apache/kafka/common/utils/MockTime.java | 24 ----
.../apache/kafka/streams/TopologyTestDriver.java | 6 -
13 files changed, 108 insertions(+), 194 deletions(-)
diff --git
a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index b310020e125..cebd325f424 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -441,8 +441,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
config.getLong(ProducerConfig.METADATA_MAX_AGE_CONFIG),
config.getLong(ProducerConfig.METADATA_MAX_IDLE_CONFIG),
logContext,
- clusterResourceListeners,
- Time.SYSTEM);
+ clusterResourceListeners);
this.metadata.bootstrap(addresses);
}
this.transactionManager = configureTransactionState(config,
logContext);
@@ -803,7 +802,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
if (versionOpt.isEmpty()) return 0L;
sender.wakeup();
try {
- metadata.awaitUpdate(versionOpt.getAsInt(), maxBlockTimeMs);
+ metadata.awaitUpdate(versionOpt.getAsInt(),
time.timer(maxBlockTimeMs));
} catch (InterruptedException e) {
throw new InterruptException(e);
}
@@ -1274,7 +1273,7 @@ public class KafkaProducer<K, V> implements Producer<K,
V> {
int version = metadata.requestUpdateForTopic(topic);
sender.wakeup();
try {
- metadata.awaitUpdate(version, remainingWaitMs);
+ metadata.awaitUpdate(version, time.timer(remainingWaitMs));
} catch (TimeoutException ex) {
// Rethrow with original maxWaitMs to prevent logging
exception with remainingWaitMs
final String errorMessage = getErrorMessage(partitionsCount,
topic, partition, maxWaitMs);
diff --git
a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java
b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java
index 6945092c143..1e0d265e58d 100644
---
a/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java
+++
b/clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerMetadata.java
@@ -18,11 +18,12 @@ package org.apache.kafka.clients.producer.internals;
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
-import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
import org.apache.kafka.common.utils.internals.LogContext;
import org.slf4j.Logger;
@@ -44,7 +45,6 @@ public class ProducerMetadata extends Metadata {
private final Map<String, Long> topics = new ConcurrentHashMap<>();
private final Set<String> newTopics = new HashSet<>();
private final Logger log;
- private final Time time;
private Map<String, Errors> errors = null;
public ProducerMetadata(long refreshBackoffMs,
@@ -52,12 +52,10 @@ public class ProducerMetadata extends Metadata {
long metadataExpireMs,
long metadataIdleMs,
LogContext logContext,
- ClusterResourceListeners clusterResourceListeners,
- Time time) {
+ ClusterResourceListeners clusterResourceListeners)
{
super(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs,
logContext, clusterResourceListeners);
this.metadataIdleMs = metadataIdleMs;
this.log = logContext.logger(ProducerMetadata.class);
- this.time = time;
}
@Override
@@ -153,14 +151,19 @@ public class ProducerMetadata extends Metadata {
/**
* 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, () -> {
+ public synchronized void awaitUpdate(final int lastVersion, final Timer
timer) throws InterruptedException {
+ while (true) {
// Throw fatal exceptions, if there are any. Recoverable topic
errors will be handled by the caller.
maybeThrowFatalException();
- return updateVersion() > lastVersion || isClosed();
- }, deadlineMs);
+ if (updateVersion() > lastVersion || isClosed())
+ break;
+
+ timer.update();
+ if (timer.isExpired())
+ throw new TimeoutException("Failed to update metadata after "
+ timer.timeoutMs() + " ms.");
+
+ wait(timer.remainingMs());
+ }
if (isClosed())
throw new KafkaException("Requested metadata update after close");
diff --git
a/clients/src/main/java/org/apache/kafka/common/utils/SystemTime.java
b/clients/src/main/java/org/apache/kafka/common/utils/SystemTime.java
index 524353507b7..f57cde65e32 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/SystemTime.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/SystemTime.java
@@ -16,10 +16,6 @@
*/
package org.apache.kafka.common.utils;
-import org.apache.kafka.common.errors.TimeoutException;
-
-import java.util.function.Supplier;
-
/**
* A time implementation that uses the system clock and sleep call. Use
`Time.SYSTEM` instead of creating an instance
* of this class.
@@ -46,22 +42,6 @@ class SystemTime implements Time {
Utils.sleep(ms);
}
- @Override
- public void waitObject(Object obj, Supplier<Boolean> condition, long
deadlineMs) throws InterruptedException {
- synchronized (obj) {
- while (true) {
- if (condition.get())
- return;
-
- long currentTimeMs = milliseconds();
- if (currentTimeMs >= deadlineMs)
- throw new TimeoutException("Condition not satisfied before
deadline");
-
- obj.wait(deadlineMs - currentTimeMs);
- }
- }
- }
-
private SystemTime() {
}
diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Time.java
b/clients/src/main/java/org/apache/kafka/common/utils/Time.java
index b2277d24bac..500277dfef7 100644
--- a/clients/src/main/java/org/apache/kafka/common/utils/Time.java
+++ b/clients/src/main/java/org/apache/kafka/common/utils/Time.java
@@ -73,7 +73,20 @@ public interface Time {
*
* @throws org.apache.kafka.common.errors.TimeoutException if the timeout
expires before the condition is satisfied
*/
- void waitObject(Object obj, Supplier<Boolean> condition, long deadlineMs)
throws InterruptedException;
+ default void waitObject(Object obj, Supplier<Boolean> condition, long
deadlineMs) throws InterruptedException {
+ synchronized (obj) {
+ while (true) {
+ if (condition.get())
+ return;
+
+ long currentTimeMs = milliseconds();
+ if (currentTimeMs >= deadlineMs)
+ throw new
org.apache.kafka.common.errors.TimeoutException("Condition not satisfied before
deadline");
+
+ obj.wait(deadlineMs - currentTimeMs);
+ }
+ }
+ }
/**
* Get a timer which is bound to this time instance and expires after the
given timeout
diff --git
a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index 628caabb0cf..d2524fd2cf0 100644
---
a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -94,6 +94,7 @@ import
org.apache.kafka.common.telemetry.internals.ClientTelemetrySender;
import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.common.utils.Timer;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.common.utils.internals.ProducerIdAndEpoch;
import org.apache.kafka.test.MockMetricsReporter;
@@ -128,7 +129,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -812,19 +812,19 @@ public class KafkaProducerTest {
// One request update for each empty cluster returned
verify(metadata, times(4)).requestUpdateForTopic(topic);
- verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(4)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(5)).fetch();
// Should not request update for subsequent `send`
producer.send(record, null);
verify(metadata, times(4)).requestUpdateForTopic(topic);
- verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(4)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(6)).fetch();
// Should not request update for subsequent `partitionsFor`
producer.partitionsFor(topic);
verify(metadata, times(4)).requestUpdateForTopic(topic);
- verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(4)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(7)).fetch();
producer.close(Duration.ofMillis(0));
@@ -846,13 +846,13 @@ public class KafkaProducerTest {
// Verify the topic's metadata isn't requested since it's already
present.
verify(metadata, times(0)).requestUpdateForTopic(topic);
- verify(metadata, times(0)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(0)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(1)).fetch();
// The metadata has been expired. Verify the producer requests the
topic's metadata.
producer.send(record, null);
verify(metadata, times(1)).requestUpdateForTopic(topic);
- verify(metadata, times(1)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(1)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(3)).fetch();
producer.close(Duration.ofMillis(0));
@@ -888,7 +888,7 @@ public class KafkaProducerTest {
// For idempotence enabled case, the first metadata.fetch will be
called in Sender#maybeSendAndPollTransactionalRequest
Future<RecordMetadata> future = producer.send(record);
verify(metadata, times(4)).requestUpdateForTopic(topic);
- verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(4)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(5)).fetch();
try {
assertInstanceOf(TimeoutException.class,
assertThrows(ExecutionException.class, future::get).getCause());
@@ -917,7 +917,7 @@ public class KafkaProducerTest {
// One request update if metadata is available but outdated for the
given record
producer.send(record);
verify(metadata, times(2)).requestUpdateForTopic(topic);
- verify(metadata, times(2)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(2)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(3)).fetch();
producer.close(Duration.ofMillis(0));
@@ -955,7 +955,7 @@ public class KafkaProducerTest {
Future<RecordMetadata> future = producer.send(record);
verify(metadata, times(4)).requestUpdateForTopic(topic);
- verify(metadata, times(4)).awaitUpdate(anyInt(), anyLong());
+ verify(metadata, times(4)).awaitUpdate(anyInt(), any(Timer.class));
verify(metadata, times(5)).fetch();
try {
assertInstanceOf(TimeoutException.class,
assertThrows(ExecutionException.class, future::get).getCause());
@@ -968,16 +968,16 @@ public class KafkaProducerTest {
public void testTopicRefreshInMetadata() throws InterruptedException {
Map<String, Object> configs = new HashMap<>();
configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
- configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "600000");
+ configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "3000");
// test under normal producer for simplicity
configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, false);
- long refreshBackoffMs = 500L;
- long refreshBackoffMaxMs = 5000L;
+ long refreshBackoffMs = 100L;
+ 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";
try (KafkaProducer<String, String> producer = kafkaProducer(configs,
new StringSerializer(), new StringSerializer(), metadata, new
MockClient(time, metadata), null, time)) {
@@ -991,7 +991,6 @@ public class KafkaProducerTest {
MetadataResponse updateResponse =
RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
singletonMap(topic,
Errors.UNKNOWN_TOPIC_OR_PARTITION), emptyMap());
metadata.updateWithCurrentRequestVersion(updateResponse,
false, time.milliseconds());
- time.sleep(60 * 1000L);
}
});
t.start();
@@ -1003,42 +1002,32 @@ public class KafkaProducerTest {
}
@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();
}
}
@@ -1046,48 +1035,42 @@ public class KafkaProducerTest {
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);
- // Update the metadata again, but it should be expired at
this point.
- updateResponse = RequestTestUtils.metadataUpdateWith(1,
singletonMap(topic, 1));
- metadata.updateWithCurrentRequestVersion(updateResponse,
false, time.milliseconds());
- exchanger.exchange(null); // 3
- while (!metadata.updateRequested())
- Thread.sleep(100);
- time.sleep(30 * 1000L);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- t.start();
- exchanger.exchange(null); // 1
- assertNotNull(producer.partitionsFor(topic));
- exchanger.exchange(null); // 2
- exchanger.exchange(null); // 3
+ // Force a metadata update so the Sender consumes the queued empty
response.
+ // This triggers retainTopic() which removes the expired topic
from the snapshot.
+ metadata.requestUpdate(true);
+ // Give the Sender time to process the update
+ Thread.sleep(500);
+
+ // partitionsFor should time out because the topic was expired and
the Sender
+ // now replays empty metadata. The real wait() timeout kicks in
after MAX_BLOCK_MS.
assertThrows(TimeoutException.class, () ->
producer.partitionsFor(topic));
- t.join();
}
}
@@ -2593,7 +2576,7 @@ public class KafkaProducerTest {
Time time = Time.SYSTEM;
MetadataResponse initialUpdateResponse =
RequestTestUtils.metadataUpdateWith(1, emptyMap());
ProducerMetadata metadata = new ProducerMetadata(0, 0, Long.MAX_VALUE,
Long.MAX_VALUE,
- new LogContext(), new ClusterResourceListeners(), time);
+ new LogContext(), new ClusterResourceListeners());
metadata.updateWithCurrentRequestVersion(initialUpdateResponse, false,
time.milliseconds());
MockClient client = new MockClient(time, metadata);
@@ -2749,7 +2732,7 @@ public class KafkaProducerTest {
private static ProducerMetadata newMetadata(long refreshBackoffMs, long
refreshBackoffMaxMs, long expirationMs) {
return new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs,
expirationMs, DEFAULT_METADATA_IDLE_MS,
- new LogContext(), new ClusterResourceListeners(), Time.SYSTEM);
+ new LogContext(), new ClusterResourceListeners());
}
@Test
@@ -2800,8 +2783,7 @@ public class KafkaProducerTest {
configs.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
MockProducerInterceptor.class.getName());
configs.put(MockProducerInterceptor.APPEND_STRING_PROP, "something");
-
- Time time = new MockTime();
+ Time time = Time.SYSTEM;
ProducerMetadata producerMetadata = newMetadata(0, 0, Long.MAX_VALUE);
MockClient client = new MockClient(time, producerMetadata);
diff --git
a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
index 01b202e29c8..559ab4d12d5 100644
---
a/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerMetadataTest.java
@@ -48,7 +48,7 @@ public class ProducerMetadataTest {
private final long refreshBackoffMaxMs = 1000;
private final long metadataExpireMs = 1000;
private final ProducerMetadata metadata = new
ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs,
METADATA_IDLE_MS,
- new LogContext(), new ClusterResourceListeners(), Time.SYSTEM);
+ new LogContext(), new ClusterResourceListeners());
private final AtomicReference<Exception> backgroundError = new
AtomicReference<>();
@AfterEach
@@ -131,7 +131,7 @@ public class ProducerMetadataTest {
}
/**
- * Tests that {@link
org.apache.kafka.clients.producer.internals.ProducerMetadata#awaitUpdate(int,
long)} doesn't
+ * Tests that {@link
org.apache.kafka.clients.producer.internals.ProducerMetadata#awaitUpdate(int,
org.apache.kafka.common.utils.Timer)} doesn't
* wait forever with a max timeout value of 0
*
* @throws Exception
@@ -144,7 +144,7 @@ public class ProducerMetadataTest {
assertTrue(metadata.timeToNextUpdate(time) > 0, "No update needed.");
// first try with a max wait time of 0 and ensure that this returns
back without waiting forever
try {
- metadata.awaitUpdate(metadata.requestUpdate(true), 0);
+ metadata.awaitUpdate(metadata.requestUpdate(true),
Time.SYSTEM.timer(0));
fail("Wait on metadata update was expected to timeout, but it
didn't");
} catch (TimeoutException te) {
// expected
@@ -152,7 +152,7 @@ public class ProducerMetadataTest {
// now try with a higher timeout value once
final long twoSecondWait = 2000;
try {
- metadata.awaitUpdate(metadata.requestUpdate(true), twoSecondWait);
+ metadata.awaitUpdate(metadata.requestUpdate(true),
Time.SYSTEM.timer(twoSecondWait));
fail("Wait on metadata update was expected to timeout, but it
didn't");
} catch (TimeoutException te) {
// expected
@@ -215,7 +215,7 @@ public class ProducerMetadataTest {
@Test
public void testMetadataWaitAbortedOnFatalException() {
metadata.fatalError(new AuthenticationException("Fatal exception from
test"));
- assertThrows(AuthenticationException.class, () ->
metadata.awaitUpdate(0, 1000));
+ assertThrows(AuthenticationException.class, () ->
metadata.awaitUpdate(0, Time.SYSTEM.timer(1000)));
}
@Test
@@ -351,7 +351,7 @@ public class ProducerMetadataTest {
Thread thread = new Thread(() -> {
try {
while (metadata.fetch().partitionsForTopic(topic).isEmpty())
- metadata.awaitUpdate(metadata.requestUpdate(false),
maxWaitMs);
+ metadata.awaitUpdate(metadata.requestUpdate(false),
Time.SYSTEM.timer(maxWaitMs));
} catch (Exception e) {
backgroundError.set(e);
}
diff --git
a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
index 3464f6e4ef0..61dcab8ef07 100644
---
a/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java
@@ -171,7 +171,7 @@ public class SenderTest {
private MockTime time = new MockTime();
private final int batchSize = 16 * 1024;
private final ProducerMetadata metadata = new ProducerMetadata(0, 0,
Long.MAX_VALUE, TOPIC_IDLE_MS,
- new LogContext(), new ClusterResourceListeners(), time);
+ new LogContext(), new ClusterResourceListeners());
private final ApiVersions apiVersions = new ApiVersions();
private MockClient client = new MockClient(time, metadata);
private Metrics metrics = null;
diff --git
a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
index 849c31fff3f..97fe1913fa8 100644
---
a/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
+++
b/clients/src/test/java/org/apache/kafka/clients/producer/internals/TransactionManagerTest.java
@@ -154,7 +154,7 @@ public class TransactionManagerTest {
private final LogContext logContext = new LogContext();
private final MockTime time = new MockTime();
private final ProducerMetadata metadata = new ProducerMetadata(0, 0,
Long.MAX_VALUE, Long.MAX_VALUE,
- logContext, new ClusterResourceListeners(), time);
+ logContext, new ClusterResourceListeners());
private final MockClient client = new MockClient(time, metadata);
private final ApiVersions apiVersions = new ApiVersions();
diff --git
a/clients/src/test/java/org/apache/kafka/common/utils/MockTimeTest.java
b/clients/src/test/java/org/apache/kafka/common/utils/MockTimeTest.java
index 88a18fe897c..784b2cadc1b 100644
--- a/clients/src/test/java/org/apache/kafka/common/utils/MockTimeTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/MockTimeTest.java
@@ -22,7 +22,7 @@ import org.junit.jupiter.api.Timeout;
import static org.junit.jupiter.api.Assertions.assertEquals;
@Timeout(120)
-public class MockTimeTest extends TimeTest {
+public class MockTimeTest {
@Test
public void testAdvanceClock() {
@@ -42,9 +42,4 @@ public class MockTimeTest extends TimeTest {
assertEquals(103, time.milliseconds());
assertEquals(104, time.milliseconds());
}
-
- @Override
- protected Time createTime() {
- return new MockTime();
- }
}
diff --git
a/clients/src/test/java/org/apache/kafka/common/utils/SystemTimeTest.java
b/clients/src/test/java/org/apache/kafka/common/utils/SystemTimeTest.java
deleted file mode 100644
index edc53d2293e..00000000000
--- a/clients/src/test/java/org/apache/kafka/common/utils/SystemTimeTest.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.common.utils;
-
-public class SystemTimeTest extends TimeTest {
-
- @Override
- protected Time createTime() {
- return Time.SYSTEM;
- }
-}
diff --git a/clients/src/test/java/org/apache/kafka/common/utils/TimeTest.java
b/clients/src/test/java/org/apache/kafka/common/utils/TimeTest.java
index 1ca630f81cd..e9d62cc3c5a 100644
--- a/clients/src/test/java/org/apache/kafka/common/utils/TimeTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/utils/TimeTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.common.utils;
import org.apache.kafka.common.errors.TimeoutException;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -27,16 +28,15 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-public abstract class TimeTest {
+@Timeout(120)
+public class TimeTest {
- protected abstract Time createTime();
+ private final Time time = Time.SYSTEM;
@Test
public void testWaitObjectTimeout() throws InterruptedException {
Object obj = new Object();
- Time time = createTime();
- long timeoutMs = 100;
- long deadlineMs = time.milliseconds() + timeoutMs;
+ long deadlineMs = time.milliseconds() + 100;
AtomicReference<Exception> caughtException = new AtomicReference<>();
Thread t = new Thread(() -> {
try {
@@ -47,7 +47,6 @@ public abstract class TimeTest {
});
t.start();
- time.sleep(timeoutMs);
t.join();
assertEquals(TimeoutException.class, caughtException.get().getClass());
@@ -56,9 +55,7 @@ public abstract class TimeTest {
@Test
public void testWaitObjectConditionSatisfied() throws InterruptedException
{
Object obj = new Object();
- Time time = createTime();
- long timeoutMs = 1000000000;
- long deadlineMs = time.milliseconds() + timeoutMs;
+ long deadlineMs = time.milliseconds() + 1000000000;
AtomicBoolean condition = new AtomicBoolean(false);
AtomicReference<Exception> caughtException = new AtomicReference<>();
Thread t = new Thread(() -> {
diff --git
a/clients/src/testFixtures/java/org/apache/kafka/common/utils/MockTime.java
b/clients/src/testFixtures/java/org/apache/kafka/common/utils/MockTime.java
index ccf3eec196e..5bb6505947c 100644
--- a/clients/src/testFixtures/java/org/apache/kafka/common/utils/MockTime.java
+++ b/clients/src/testFixtures/java/org/apache/kafka/common/utils/MockTime.java
@@ -16,12 +16,9 @@
*/
package org.apache.kafka.common.utils;
-import org.apache.kafka.common.errors.TimeoutException;
-
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
/**
* A clock that you can manually advance by calling sleep
@@ -86,27 +83,6 @@ public class MockTime implements Time {
tick();
}
- @Override
- public void waitObject(Object obj, Supplier<Boolean> condition, long
deadlineMs) throws InterruptedException {
- Listener listener = () -> {
- synchronized (obj) {
- obj.notify();
- }
- };
- listeners.add(listener);
- try {
- synchronized (obj) {
- while (milliseconds() < deadlineMs && !condition.get()) {
- obj.wait();
- }
- if (!condition.get())
- throw new TimeoutException("Condition not satisfied before
deadline");
- }
- } finally {
- listeners.remove(listener);
- }
- }
-
public void setCurrentTimeMs(long newMs) {
long oldMs = timeMs.getAndSet(newMs);
diff --git
a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
index 54020c50de9..f491c03fb86 100644
---
a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
+++
b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -125,7 +125,6 @@ import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
import java.util.regex.Pattern;
import static
org.apache.kafka.streams.internals.StreamsConfigUtils.ProcessingMode.EXACTLY_ONCE_V2;
@@ -1361,11 +1360,6 @@ public class TopologyTestDriver implements Closeable {
timeMs.addAndGet(ms);
highResTimeNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(ms));
}
-
- @Override
- public void waitObject(final Object obj, final Supplier<Boolean>
condition, final long timeoutMs) {
- throw new UnsupportedOperationException();
- }
}
static class KeyValueStoreFacade<K, V> extends
GenericReadOnlyKeyValueStoreFacade<K, ValueAndTimestamp<V>, V> implements
KeyValueStore<K, V> {
