AndrewJSchofield commented on code in PR #15455:
URL: https://github.com/apache/kafka/pull/15455#discussion_r1510342872
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java:
##########
@@ -153,15 +154,17 @@ public void testMetadataUpdateEvent() {
@Test
public void testAsyncCommitEvent() {
- ApplicationEvent e = new AsyncCommitEvent(new HashMap<>());
+ Timer timer = time.timer(100);
Review Comment:
This should be `Long.MAX_VALUE` I think.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java:
##########
@@ -170,7 +173,8 @@ public void testSyncCommitEvent() {
@Test
public void testListOffsetsEventIsProcessed() {
Map<TopicPartition, Long> timestamps = Collections.singletonMap(new
TopicPartition("topic1", 1), 5L);
- ApplicationEvent e = new ListOffsetsEvent(timestamps, true);
+ Timer timer = time.timer(100);
Review Comment:
And again. I don't think you're actually trying to bound the event's time
here, but you have to provide a timer. I reckon `Long.MAX_VALUE` is better
aligned with the spirit of this event.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitEvent.java:
##########
@@ -29,8 +30,8 @@ public abstract class CommitEvent extends
CompletableApplicationEvent<Void> {
*/
private final Map<TopicPartition, OffsetAndMetadata> offsets;
- protected CommitEvent(final Type type, final Map<TopicPartition,
OffsetAndMetadata> offsets) {
- super(type);
+ protected CommitEvent(final Type type, final Timer timer, final
Map<TopicPartition, OffsetAndMetadata> offsets) {
Review Comment:
A nit perhaps but you've put the Timer as the final argument in all other
cases, so I suggest you do it for this one too for consistency.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java:
##########
@@ -283,8 +291,9 @@ void testEnsureEventsAreCompleted() {
coordinatorRequestManager.markCoordinatorUnknown("test",
time.milliseconds());
client.prepareResponse(FindCoordinatorResponse.prepareResponse(Errors.NONE,
"group-id", node));
prepareOffsetCommitRequest(new HashMap<>(), Errors.NONE, false);
- CompletableApplicationEvent<Void> event1 = spy(new
AsyncCommitEvent(Collections.emptyMap()));
- ApplicationEvent event2 = new AsyncCommitEvent(Collections.emptyMap());
+ Timer timer = time.timer(100);
Review Comment:
And it continues.
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThreadTest.java:
##########
@@ -199,7 +205,8 @@ public void testResetPositionsProcessFailureIsIgnored() {
@Test
public void testValidatePositionsEventIsProcessed() {
- ValidatePositionsEvent e = new ValidatePositionsEvent();
+ Timer timer = time.timer(100);
Review Comment:
And here.
--
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
