Skip to content

test whether it is possible to consume before the producer gets an ack #97

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

Open
arov00 wants to merge 2 commits into
base: main
Choose a base branch
from
+20 −9
Open

test whether it is possible to consume before the producer gets an ack #97

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8d5865b
test whether it is possible to consume before the producer gets an ack
arov00 Mar 3, 2026
bc27353
more consistent timing
arov00 Mar 3, 2026
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
19 changes: 15 additions & 4 deletions src/main/java/io/spoud/MetricService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import io.quarkus.logging.Log;
import io.spoud.config.SynthClientConfig;
import io.spoud.kafka.PartitionRebalancer;
import io.vertx.core.impl.ConcurrentHashSet;
import jakarta.enterprise.context.ApplicationScoped;
import org.eclipse.microprofile.config.inject.ConfigProperty;

Expand Down Expand Up @@ -89,7 +90,12 @@ public void recordProducedFailure() {
recordsFailedCounter.increment();
}

synchronized public void recordLatency(String topic, int partition, long latencyMs, String fromRack) {
synchronized public void recordLatency(String topic, int partition, long latencyMs, String fromRack, long offset) {
if (fromRack.equals(config.rack()) && !ackedOffsets.contains(new AckedMessage(offset, partition))) {
Log.warnf("Consumed message %d:%d before receiving an ACK! This means that Ack latency could be higher than E2E latency!", partition, offset);
} else {
ackedOffsets.remove(new AckedMessage(offset, partition));
}
Comment on lines +94 to +98

@spike83 spike83 Mar 3, 2026

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.

here we cause a potential memory leak as we never remove the item that was missing here if it's added to the list later on. list will grow forever.

Log.debugv("Latency for partition {0}: {1}ms", partition, latencyMs);
if (partitionRebalancer.isInitialRefreshPending()) {
Log.info("Ignoring latencies as the initial partition assignment is not done yet");
Expand Down Expand Up @@ -132,7 +138,12 @@ public Collection<WrappedDistributionSummary> getE2ELatencies() {
return e2eLatencies.values();
}

public void recordAckLatency(String topic, int partition, Duration between) {
public record AckedMessage(long offset, int partition) {
}
private final Set<AckedMessage> ackedOffsets = new ConcurrentHashSet<>();

public void recordAckLatency(String topic, int partition, Duration between, long offset) {
ackedOffsets.add(new AckedMessage(offset, partition));
Log.debugv("Ack latency for partition {0}: {1}ms", partition, between.toMillis());
if (partitionRebalancer.isInitialRefreshPending()) {
Log.info("Ignoring ack latency as the initial partition assignment is not done yet");
Expand Down Expand Up @@ -192,8 +203,8 @@ private WrappedDistributionSummary genE2eSummary(String topic, int partition, St
.tag(TAG_FROM_RACK, fromRack)
.tag(TAG_BROKER_RACK, brokerRack)
.description("End-to-end latency of the synthetic client")
.minimumExpectedValue(1.0)
.maximumExpectedValue(10_000.0)
.minimumExpectedValue(config.expectedMinLatency())
.maximumExpectedValue(config.expectedMaxLatency())

@spike83 spike83 Mar 3, 2026

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.

this makes sense. might be influencing the P9s to a certain extend. This change should survive.

.publishPercentiles(0.5, 0.8, 0.9, 0.95, 0.99)
.publishPercentileHistogram(config.publishHistogramBuckets())
.distributionStatisticExpiry(config.samplingTimeWindow())
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/io/spoud/kafka/MessageConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ public void onPartitionsAssigned(Collection<TopicPartition> collection) {
.map(String::new)
.orElse(null);
metricService.recordConsumptionTime();
metricService.recordLatency(message.topic(), message.partition(), consumeTime - produceTime, fromRack);
metricService.recordLatency(message.topic(), message.partition(), consumeTime - produceTime, fromRack, message.offset());
advertisedListenerRepository.mapRackToUrl(fromRack, advertisedListener);
lastReport.updateAndGet(last -> {
if (Duration.between(last, Instant.now()).getSeconds() > 10) {
Expand Down
8 changes: 4 additions & 4 deletions src/main/java/io/spoud/kafka/MessageProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,6 @@ public void recreateProducer() {
}

public void send(Long key, String value) {
Instant send = Instant.now();
var record = new ProducerRecord<>(config.topic(), null, timeService.currentTimeMillis(), key, value);
record.headers().add(HEADER_RACK, config.rack().getBytes());
record.headers().add(HEADER_ADVERTISED_LISTENER, config.advertisedListener().orElse("").getBytes());
Expand All @@ -87,9 +86,10 @@ var record = new ProducerRecord<>(config.topic(), null, timeService.currentTimeM
Log.error("Failed to send message", exception);
metricService.recordProducedFailure();
} else {
Instant ack = Instant.now();
lastMessage.set(ack);
metricService.recordAckLatency(metadata.topic(), metadata.partition(), Duration.between(send, ack));
Instant ackTime = Instant.ofEpochMilli(timeService.currentTimeMillis());
Instant sendTime = Instant.ofEpochMilli(metadata.timestamp());

@spike83 spike83 Mar 3, 2026

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.

metadata.timestamp is now set by broker? this would add another time source into the mix, no?

@arov00 arov00 Mar 3, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

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

I think metadata.timestamp is the timestamp we give during production, no?

@spike83 spike83 Mar 4, 2026

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.

yes, depending on the topic configuration log.message.timestamp.type or message.timestamp.type.
If we are unlucky we inherit the LogAppendTime timestamp from broker config as a default when creating topics. We should ensure we set/validate CreateTime on the topics.

lastMessage.set(ackTime);
metricService.recordAckLatency(metadata.topic(), metadata.partition(), Duration.between(sendTime, ackTime), metadata.offset());
metricService.recordProducedSuccess();
}
});
Expand Down