Skip to content

MINOR: fix time discrepancy between TestInputTopic and TopologyTestDriver that creates it #17702

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ableegoldman merged 1 commit into from
Nov 9, 2024
Merged

MINOR: fix time discrepancy between TestInputTopic and TopologyTestDriver that creates it #17702

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ private static boolean isRecoverable(final KafkaException uncaughtException) {
* @throws IllegalStateException if EOS is disabled
* @throws TaskMigratedException
*/
protected void commitTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets,
public void commitTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets,

@ableegoldman ableegoldman Nov 6, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

unrelated to the main fix, just doing some cleanup of the TTD on the side. I don't see why we can't just make this method public since it's an internal class anyways -- which was the exact reasoning I saw on the PR which originally changed it from private to protected anyway. But I can leave this part out if there are any concerns

final ConsumerGroupMetadata consumerGroupMetadata) {
if (!eosEnabled()) {
throw new IllegalStateException(formatException("Exactly-once is not enabled"));
Expand Down
35 changes: 13 additions & 22 deletions streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,6 @@
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.metrics.MetricConfig;
import org.apache.kafka.common.metrics.Metrics;
Expand All @@ -44,7 +43,6 @@
import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
import org.apache.kafka.streams.errors.TopologyException;
import org.apache.kafka.streams.internals.StreamsConfigUtils;
import org.apache.kafka.streams.internals.StreamsConfigUtils.ProcessingMode;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
Expand Down Expand Up @@ -233,7 +231,7 @@ public class TopologyTestDriver implements Closeable {

private final MockConsumer<byte[], byte[]> consumer;
private final MockProducer<byte[], byte[]> producer;
private final TestDriverProducer testDriverProducer;
private final StreamsProducer testDriverProducer;

private final Map<String, TopicPartition> partitionsByInputTopic = new HashMap<>();
private final Map<String, TopicPartition> globalPartitionsByInputTopic = new HashMap<>();
Expand Down Expand Up @@ -345,7 +343,8 @@ public List<PartitionInfo> partitionsFor(final String topic) {
return Collections.singletonList(new PartitionInfo(topic, PARTITION_ID, null, null, null));
}
};
testDriverProducer = new TestDriverProducer(

testDriverProducer = new StreamsProducer(
producer,
StreamsConfigUtils.processingMode(streamsConfig),
mockWallClockTime,
Expand Down Expand Up @@ -738,7 +737,14 @@ private Queue<ProducerRecord<byte[], byte[]>> getRecordsQueue(final String topic
public final <K, V> TestInputTopic<K, V> createInputTopic(final String topicName,

@ableegoldman ableegoldman Nov 6, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also wondering why all these methods are final -- I could have addressed this issue in our code without this PR if I could just override these methods. Are we concerned about users doing this..? And why?

I figure this change actually would need a KIP so I left it out, but I'd like to remove the final from these unless someone knows a good reason for doing this

@lucasbru lucasbru Nov 6, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right, you'd need a KIP for removing it. I'm fine with merging this fix without removing the final.

For removing the final -- it does make sense to restrict the surface area for extending the public interface. I'm not sure why we'd need to override those. But this is something for the KIP discussion anyways, if you want to write one

final Serializer<K> keySerializer,
final Serializer<V> valueSerializer) {
return new TestInputTopic<>(this, topicName, keySerializer, valueSerializer, Instant.now(), Duration.ZERO);
return new TestInputTopic<>(
this,
topicName,
keySerializer,
valueSerializer,
Instant.ofEpochMilli(mockWallClockTime.milliseconds()),

@ableegoldman ableegoldman Nov 6, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the main fix: instead of initializing to the current time, we should initialize to the driver's current time. I feel like this makes more sense anyway...we mock time in the driver so why should the TestInputTopic use a view of time based on actual wallclock time rather than the driver's time?

And would this need a KIP?

@lucasbru lucasbru Nov 6, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think it needs a KIP

@bbejeck bbejeck Nov 6, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I agree this doesn't need a KIP, plus there's an overload where users can provide their own time if needed. Although, you make a great point about synchronizing the time with the driver.

Duration.ZERO
);
}

/**
Expand Down Expand Up @@ -985,7 +991,7 @@ private void throwIfBuiltInStore(final StateStore stateStore) {
public <K, V> KeyValueStore<K, V> getKeyValueStore(final String name) {
final StateStore store = getStateStore(name, false);
if (store instanceof TimestampedKeyValueStore) {
log.info("Method #getTimestampedKeyValueStore() should be used to access a TimestampedKeyValueStore.");
log.warn("Method #getTimestampedKeyValueStore() should be used to access a TimestampedKeyValueStore.");
return new KeyValueStoreFacade<>((TimestampedKeyValueStore<K, V>) store);
}
return store instanceof KeyValueStore ? (KeyValueStore<K, V>) store : null;
Expand Down Expand Up @@ -1063,7 +1069,7 @@ public <K, V> VersionedKeyValueStore<K, V> getVersionedKeyValueStore(final Strin
public <K, V> WindowStore<K, V> getWindowStore(final String name) {
final StateStore store = getStateStore(name, false);
if (store instanceof TimestampedWindowStore) {
log.info("Method #getTimestampedWindowStore() should be used to access a TimestampedWindowStore.");
log.warn("Method #getTimestampedWindowStore() should be used to access a TimestampedWindowStore.");
return new WindowStoreFacade<>((TimestampedWindowStore<K, V>) store);
}
return store instanceof WindowStore ? (WindowStore<K, V>) store : null;
Expand Down Expand Up @@ -1351,19 +1357,4 @@ public Position getPosition() {
}
}

private static class TestDriverProducer extends StreamsProducer {

@ableegoldman ableegoldman Nov 6, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This class isn't doing anything, right? I'm unclear on its purpose and figured I might as well clean it up from the TTD -- but again, happy to undo this change since it's not needed for the main fix

@lucasbru lucasbru Nov 6, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interesting... seems fine to me

@ableegoldman ableegoldman Nov 6, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This class isn't doing anything, right? I'm unclear on its purpose and figured I might as well clean it up from the TTD -- but again, happy to undo this change since it's not needed for the main fix


public TestDriverProducer(final Producer<byte[], byte[]> producer,
final ProcessingMode processingMode,
final Time time,
final LogContext logContext) {
super(producer, processingMode, time, logContext);
}

@Override
public void commitTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets,
final ConsumerGroupMetadata consumerGroupMetadata) throws ProducerFencedException {
super.commitTransaction(offsets, consumerGroupMetadata);
}
}
}