You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "philipnee (via GitHub)" <gi...@apache.org> on 2023/07/28 17:24:06 UTC
[GitHub] [kafka] philipnee opened a new pull request, #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
philipnee opened a new pull request, #14123:
URL: https://github.com/apache/kafka/pull/14123
**Summary**
I discovered the committed() API timeout during the integration test. After investigation, this is because the future was not completed in the ApplicationEventProcessor. I also added `toString` methods to the event class for debug purposes.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1659084728
Current failures:
```
New failing - 12
Build / JDK 11 and Scala 2.13 / [1] tlsProtocol=TLSv1.2, useInlinePem=false – org.apache.kafka.common.network.SslTransportLayerTest
<1s
Build / JDK 11 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest
1m 45s
Build / JDK 20 and Scala 2.13 / testNonDefaultConnectionCountLimitAndRateLimit() – kafka.network.ConnectionQuotasTest
10s
Build / JDK 20 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
Build / JDK 17 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest
1m 51s
Build / JDK 17 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsWithCustomForwardingAdminIntegrationTest
1m 43s
Build / JDK 8 and Scala 2.12 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.IdentityReplicationIntegrationTest
2m 6s
Build / JDK 8 and Scala 2.12 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
2m 7s
Build / JDK 8 and Scala 2.12 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest
2m 28s
Build / JDK 8 and Scala 2.12 / testRackAwareRangeAssignor() – integration.kafka.server.FetchFromFollowerIntegrationTest
41s
Build / JDK 8 and Scala 2.12 / testClose() – kafka.log.remote.RemoteIndexCacheTest
<1s
Build / JDK 8 and Scala 2.12 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
Existing failures - 4
Build / JDK 11 and Scala 2.13 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
1m 48s
Build / JDK 20 and Scala 2.13 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
1m 48s
Build / JDK 20 and Scala 2.13 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
2m 44s
Build / JDK 8 and Scala 2.12 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
2m 18s
```
Non related to the code change. No build failure.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1281412776
##########
core/src/test/scala/integration/kafka/api/BaseAsyncConsumerTest.scala:
##########
@@ -35,8 +39,12 @@ class BaseAsyncConsumerTest extends AbstractConsumerTest {
consumer.commitAsync(cb)
waitUntilTrue(() => {
cb.successCount == 1
- }, "wait until commit is completed successfully", 5000)
+ }, "wait until commit is completed successfully", defaultBlockingAPITimeoutMs)
+ val committedOffset = consumer.committed(Set(tp).asJava, Duration.ofMillis(defaultBlockingAPITimeoutMs))
+
assertTrue(consumer.assignment.contains(tp))
+ assertNotNull(committedOffset)
+ assertNull(committedOffset.get(tp))
Review Comment:
In this test, we aren't committing any offset, so the actual return of the commit response looks like this:
(same for both old and the async one) `{topic-0=null}`
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1281188661
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventProcessor.java:
##########
@@ -106,12 +106,18 @@ private boolean process(final CommitApplicationEvent event) {
private boolean process(final OffsetFetchApplicationEvent event) {
Optional<RequestManager> commitRequestManger = registry.get(RequestManager.Type.COMMIT);
if (!commitRequestManger.isPresent()) {
- event.future.completeExceptionally(new KafkaException("Unable to fetch committed offset because the " +
+ event.future().completeExceptionally(new KafkaException("Unable to fetch committed offset because the " +
Review Comment:
Would it be better to call `event.future()` once and reuse the result?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/OffsetFetchApplicationEvent.java:
##########
@@ -19,25 +19,33 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
-import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
public class OffsetFetchApplicationEvent extends ApplicationEvent {
- final public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
- public final Set<TopicPartition> partitions;
+ private final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+ private final Set<TopicPartition> partitions;
public OffsetFetchApplicationEvent(final Set<TopicPartition> partitions) {
super(Type.FETCH_COMMITTED_OFFSET);
this.partitions = partitions;
this.future = new CompletableFuture<>();
}
- public Map<TopicPartition, OffsetAndMetadata> complete(final Duration duration) throws ExecutionException, InterruptedException, TimeoutException {
- return future.get(duration.toMillis(), TimeUnit.MILLISECONDS);
+ public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future() {
+ return this.future;
Review Comment:
this.future => future
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -150,15 +151,33 @@ public void testCommitAsync_UserSuppliedCallback() {
}
@Test
- @SuppressWarnings("unchecked")
public void testCommitted() {
Set<TopicPartition> mockTopicPartitions = mockTopicPartitionOffset().keySet();
- mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
- when(mock.complete(any())).thenReturn(new HashMap<>());
- });
- consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
- assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1)));
- verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ try (MockedConstruction<OffsetFetchApplicationEvent> mockConstruction =
Review Comment:
`mockConstruction` is unused. Also, why do we wrap under `try`? Do we expect `mockConstruction` to be closed in the end?
##########
core/src/test/scala/integration/kafka/api/BaseAsyncConsumerTest.scala:
##########
@@ -35,8 +39,12 @@ class BaseAsyncConsumerTest extends AbstractConsumerTest {
consumer.commitAsync(cb)
waitUntilTrue(() => {
cb.successCount == 1
- }, "wait until commit is completed successfully", 5000)
+ }, "wait until commit is completed successfully", defaultBlockingAPITimeoutMs)
+ val committedOffset = consumer.committed(Set(tp).asJava, Duration.ofMillis(defaultBlockingAPITimeoutMs))
+
assertTrue(consumer.assignment.contains(tp))
+ assertNotNull(committedOffset)
+ assertNull(committedOffset.get(tp))
Review Comment:
Hmm, why would the committed offset be null for tp?
##########
core/src/test/scala/integration/kafka/api/BaseAsyncConsumerTest.scala:
##########
@@ -17,12 +17,16 @@
package kafka.api
import kafka.utils.TestUtils.waitUntilTrue
-import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Assertions.{assertNotNull, assertNull, assertTrue}
import org.junit.jupiter.api.Test
+import java.time.Duration
import scala.jdk.CollectionConverters._
+
Review Comment:
extra line
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1660477317
latest test failures are unrelated
```
New failing - 10
Build / JDK 17 and Scala 2.13 / testBumpTransactionalEpoch(String).quorum=kraft – kafka.api.TransactionsTest
1m 17s
Build / JDK 17 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 17 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
14s
Build / JDK 17 and Scala 2.13 / [5] Type=Raft-Combined, Name=testDescribeQuorumStatusSuccessful, MetadataVersion=3.6-IV0, Security=PLAINTEXT – org.apache.kafka.tools.MetadataQuorumCommandTest
1m 6s
Build / JDK 20 and Scala 2.13 / testTopicIdPersistsThroughControllerRestart() – kafka.controller.ControllerIntegrationTest
5s
Build / JDK 20 and Scala 2.13 / [1] true – org.apache.kafka.streams.integration.SmokeTestDriverIntegrationTest
10m 0s
Build / JDK 8 and Scala 2.12 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest
4m 12s
Build / JDK 8 and Scala 2.12 / [1] Type=Raft-Isolated, Name=testAllocateProducersIdSentToNonController, MetadataVersion=3.6-IV0, Security=PLAINTEXT – unit.kafka.server.AllocateProducerIdsRequestTest
1m 9s
Build / JDK 8 and Scala 2.12 / [1] Type=Raft-Isolated, Name=testAllocateProducersIdSentToNonController, MetadataVersion=3.6-IV0, Security=PLAINTEXT – unit.kafka.server.AllocateProducerIdsRequestTest
1m 10s
Build / JDK 8 and Scala 2.12 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1284689796
##########
core/src/test/scala/integration/kafka/api/BaseAsyncConsumerTest.scala:
##########
@@ -35,8 +39,12 @@ class BaseAsyncConsumerTest extends AbstractConsumerTest {
consumer.commitAsync(cb)
waitUntilTrue(() => {
cb.successCount == 1
- }, "wait until commit is completed successfully", 5000)
+ }, "wait until commit is completed successfully", defaultBlockingAPITimeoutMs)
+ val committedOffset = consumer.committed(Set(tp).asJava, Duration.ofMillis(defaultBlockingAPITimeoutMs))
+
assertTrue(consumer.assignment.contains(tp))
+ assertNotNull(committedOffset)
+ assertNull(committedOffset.get(tp))
Review Comment:
Since this test is for no-consumption then the `sendRecords` could be removed in both tests I guess
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1667163104
Hey @junrao - The most recent build, after merging trunk in, seems reasonably clean. Here are the broken tests, but all seem unrelated to the change
```
Build / JDK 11 and Scala 2.13 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
1m 25s
Build / JDK 11 and Scala 2.13 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
1m 33s
Build / JDK 11 and Scala 2.13 / testSimultaneousUpwardAndDownwardDelegating – org.apache.kafka.connect.runtime.isolation.SynchronizationTest
11s
Build / JDK 20 and Scala 2.13 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
1m 40s
Build / JDK 20 and Scala 2.13 / testAutoCreateTopic(String).quorum=zk – kafka.api.PlaintextProducerSendTest
2s
Build / JDK 20 and Scala 2.13 / testAlterAndDescribe(String).quorum=kraft – kafka.server.AlterUserScramCredentialsRequestTest
1m 12s
Build / JDK 20 and Scala 2.13 / shouldRestoreAfterJoinRestart[ON_WINDOW_UPDATE_cache:true] – org.apache.kafka.streams.integration.SlidingWindowedKStreamIntegrationTest
1m 13s
Build / JDK 20 and Scala 2.13 / shouldRestoreAfterJoinRestart[ON_WINDOW_UPDATE_cache:false] – org.apache.kafka.streams.integration.SlidingWindowedKStreamIntegrationTest
1m 32s
Build / JDK 20 and Scala 2.13 / shouldRestoreAfterJoinRestart[ON_WINDOW_CLOSE_cache:true] – org.apache.kafka.streams.integration.SlidingWindowedKStreamIntegrationTest
1m 28s
Build / JDK 20 and Scala 2.13 / shouldRestoreAfterJoinRestart[ON_WINDOW_CLOSE_cache:false] – org.apache.kafka.streams.integration.SlidingWindowedKStreamIntegrationTest
1m 50s
Build / JDK 8 and Scala 2.12 / pollException() – kafka.network.SocketServerTest
16s
Build / JDK 8 and Scala 2.12 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 17 and Scala 2.13 / testReplicateSourceDefault() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationTransactionsTest
2m 18s
Build / JDK 17 and Scala 2.13 / testSimultaneousUpwardAndDownwardDelegating – org.apache.kafka.connect.runtime.isolation.SynchronizationTest
19s
Build / JDK 17 and Scala 2.13 / testCancellation(String).quorum=kraft – kafka.admin.ReassignPartitionsIntegrationTest
12s
Build / JDK 17 and Scala 2.13 / testPreferredReplica(String).quorum=kraft – kafka.server.MetadataRequestTest
8s
Build / JDK 17 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 17 and Scala 2.13 / shouldThrowIllegalArgumentExceptionWhenCustomPartitionerReturnsMultiplePartitions() – org.apache.kafka.streams.integration.KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest
1m 40s
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] lianetm commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1279887573
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -153,12 +154,32 @@ public void testCommitAsync_UserSuppliedCallback() {
@SuppressWarnings("unchecked")
public void testCommitted() {
Set<TopicPartition> mockTopicPartitions = mockTopicPartitionOffset().keySet();
- mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
- when(mock.complete(any())).thenReturn(new HashMap<>());
- });
- consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
- assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1)));
- verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ try (MockedConstruction<OffsetFetchApplicationEvent> mockConstruction =
+ mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
+ when(mock.future()).thenReturn(committedFuture);
+ })) {
+ committedFuture.complete(mockTopicPartitionOffset());
+ consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
+ assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1000)));
+ verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
Review Comment:
Does not seem needed I would say
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1656345210
We don't. Do you know when was is introduced? Is it in your inflight PR?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao merged pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #14123:
URL: https://github.com/apache/kafka/pull/14123
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1277835494
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/OffsetFetchApplicationEvent.java:
##########
@@ -19,25 +19,33 @@
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
-import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
public class OffsetFetchApplicationEvent extends ApplicationEvent {
- final public CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
- public final Set<TopicPartition> partitions;
+ private final CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future;
+ private final Set<TopicPartition> partitions;
public OffsetFetchApplicationEvent(final Set<TopicPartition> partitions) {
super(Type.FETCH_COMMITTED_OFFSET);
this.partitions = partitions;
this.future = new CompletableFuture<>();
}
- public Map<TopicPartition, OffsetAndMetadata> complete(final Duration duration) throws ExecutionException, InterruptedException, TimeoutException {
Review Comment:
Removed because this method was rather pointless
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1277840446
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -153,12 +153,14 @@ public void testCommitAsync_UserSuppliedCallback() {
@SuppressWarnings("unchecked")
public void testCommitted() {
Set<TopicPartition> mockTopicPartitions = mockTopicPartitionOffset().keySet();
- mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
- when(mock.complete(any())).thenReturn(new HashMap<>());
- });
- consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
- assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1)));
- verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ try (MockedConstruction<OffsetFetchApplicationEvent> mockConstruction = mockConstruction(OffsetFetchApplicationEvent.class,
Review Comment:
try-with-resources ensure thread-lock mockConstruction is closed to ensure other tests would function properly.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1279890789
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -153,12 +154,32 @@ public void testCommitAsync_UserSuppliedCallback() {
@SuppressWarnings("unchecked")
public void testCommitted() {
Set<TopicPartition> mockTopicPartitions = mockTopicPartitionOffset().keySet();
- mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
- when(mock.complete(any())).thenReturn(new HashMap<>());
- });
- consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
- assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1)));
- verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ try (MockedConstruction<OffsetFetchApplicationEvent> mockConstruction =
+ mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
+ when(mock.future()).thenReturn(committedFuture);
+ })) {
+ committedFuture.complete(mockTopicPartitionOffset());
+ consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
+ assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1000)));
+ verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ }
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
Review Comment:
you are 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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1661525036
@junrao - Thank you so much for the reviews. The latest commit addressed your comments. Let me know if you have more.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1659400843
Here are the most recent failures:
```
Build / JDK 11 and Scala 2.13 / testClose() – kafka.log.remote.RemoteIndexCacheTest
<1s
Build / JDK 11 and Scala 2.13 / testDelayedConfigurationOperations() – org.apache.kafka.controller.QuorumControllerTest
<1s
Build / JDK 20 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
13s
Build / JDK 8 and Scala 2.12 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
1m 48s
Build / JDK 8 and Scala 2.12 / testOffsetTranslationBehindReplicationFlow() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationExactlyOnceTest
2m 20s
```
The failures are unrelated.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1663201841
There seems to be some weirdness to the build - jdk 17 was timedout or something. Retriggered the build...
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on a diff in pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on code in PR #14123:
URL: https://github.com/apache/kafka/pull/14123#discussion_r1281407606
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumerTest.java:
##########
@@ -150,15 +151,33 @@ public void testCommitAsync_UserSuppliedCallback() {
}
@Test
- @SuppressWarnings("unchecked")
public void testCommitted() {
Set<TopicPartition> mockTopicPartitions = mockTopicPartitionOffset().keySet();
- mockConstruction(OffsetFetchApplicationEvent.class, (mock, ctx) -> {
- when(mock.complete(any())).thenReturn(new HashMap<>());
- });
- consumer = newConsumer(time, new StringDeserializer(), new StringDeserializer());
- assertDoesNotThrow(() -> consumer.committed(mockTopicPartitions, Duration.ofMillis(1)));
- verify(eventHandler).add(ArgumentMatchers.isA(OffsetFetchApplicationEvent.class));
+ CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> committedFuture = new CompletableFuture<>();
+ try (MockedConstruction<OffsetFetchApplicationEvent> mockConstruction =
Review Comment:
We need to close the mockConstruction object in the end because it is a thread-lock controller. If we don't close it, it would remain active and causes undesirable behavior for other tests. I'm using mockConstruction to mock the construction of the event object, are there better ways to mimic/mock the behavior the event construction and method calls there?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] philipnee commented on pull request #14123: MINOR: Fix committed API in the PrototypeAsyncConsumer timeout
Posted by "philipnee (via GitHub)" <gi...@apache.org>.
philipnee commented on PR #14123:
URL: https://github.com/apache/kafka/pull/14123#issuecomment-1656346526
Umm build failures for jdk 8 plus some non-related test failures:
```
Build / JDK 20 and Scala 2.13 / testSyncTopicConfigs() – org.apache.kafka.connect.mirror.integration.MirrorConnectorsIntegrationBaseTest
1m 31s
Build / JDK 11 and Scala 2.13 / testGracefulClose() – org.apache.kafka.clients.consumer.KafkaConsumerTest
2s
Build / JDK 11 and Scala 2.13 / testCacheEntryExpiry() – kafka.log.remote.RemoteIndexCacheTest
<1s
Build / JDK 17 and Scala 2.13 / testGracefulClose() – org.apache.kafka.clients.consumer.KafkaConsumerTest
2s
Build / JDK 17 and Scala 2.13 / testBalancePartitionLeaders() – org.apache.kafka.controller.QuorumControllerTest
12s
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org