Introduce origin event tracking mechanism #424
There are no files selected for viewing
Submodule opentelemetry-proto
updated
2 files
| +20 −0 | .github/workflows/fossa.yml | |
| +7 −3 | docs/specification.md |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| * Copyright 2024 Responsive Computing, Inc. | ||
agavra marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| * | ||
| * This source code is licensed under the Responsive Business Source License Agreement v1.0 | ||
| * available at: | ||
| * | ||
| * https://www.responsive.dev/legal/responsive-bsl-10 | ||
| * | ||
| * This software requires a valid Commercial License Key for production use. Trial and commercial | ||
| * licenses can be obtained at https://www.responsive.dev | ||
| */ | ||
|
|
||
| package dev.responsive.kafka.internal.clients; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.Map; | ||
| import org.apache.kafka.clients.consumer.ConsumerRecords; | ||
| import org.apache.kafka.clients.consumer.OffsetAndMetadata; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
| import org.apache.kafka.common.TopicPartition; | ||
|
|
||
| public class NoopOriginEventRecorder implements OriginEventRecorder { | ||
| @Override | ||
| public <K, V> ConsumerRecords<K, V> onPoll(final ConsumerRecords<K, V> records) { | ||
| return records; | ||
| } | ||
|
|
||
| @Override | ||
| public <K, V> ProducerRecord<K, V> onSend(final ProducerRecord<K, V> record) { | ||
| return record; | ||
| } | ||
|
|
||
| @Override | ||
| public void onConsumerCommit(final Map<TopicPartition, OffsetAndMetadata> offsets) { | ||
| } | ||
|
|
||
| @Override | ||
| public void onProducerCommit() { | ||
| } | ||
|
|
||
| @Override | ||
| public void onSendOffsetsToTransaction( | ||
| final Map<TopicPartition, OffsetAndMetadata> offsets, | ||
| final String consumerGroupId | ||
| ) { | ||
| } | ||
|
|
||
| @Override | ||
| public void onPartitionsLost(final Collection<TopicPartition> partitions) { | ||
| } | ||
|
|
||
| @Override | ||
| public void onUnsubscribe() { | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| /* | ||
| * Copyright 2025 Responsive Computing, Inc. | ||
| * | ||
| * This source code is licensed under the Responsive Business Source License Agreement v1.0 | ||
| * available at: | ||
| * | ||
| * https://www.responsive.dev/legal/responsive-bsl-10 | ||
| * | ||
| * This software requires a valid Commercial License Key for production use. Trial and commercial | ||
| * licenses can be obtained at https://www.responsive.dev | ||
| */ | ||
|
|
||
| package dev.responsive.kafka.internal.clients; | ||
|
|
||
| import java.util.BitSet; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * This class allows us to efficiently count the number of events | ||
| * between two offsets that match a certain condition. This is somewhat | ||
| * memory efficient in that we can track 100K offsets with ~1.5K longs | ||
| * (64 bits per long), or roughly 12KB. | ||
|
Comment on lines
+19
to
+23
Contributor
There was a problem hiding this comment. I'm a little confused by the point of this class. This is just counting the number of origin events since the last commit, right? It doesn't seem like we ever expose what the actual origin event offsets are, so why do we need to save them in this Lmk if you'd rather sync online because I'm sure I'm missing something here 🙂
Contributor
Author
There was a problem hiding this comment. This class should definitely have javadoc explaining it, originally I had implemented what you're saying but Rohan pointed out these two concerns: #424 (comment) and #424 (comment) both fixed by this bitset tracker.
Contributor
There was a problem hiding this comment. ah, I missed those discussions, but yeah he's right. thanks for filling in the javadocs |
||
| * <p> | ||
| * This is necessary for origin event tracking: since we mark records | ||
| * on poll but report aligned to commit it is possible that there are | ||
| * records that are polled but not included in the committed offsets. | ||
| * This class remedies that by allowing us to count the marked events | ||
| * up to the committed offset. | ||
| * <p> | ||
| * Additionally, there are some situations that would cause us to poll | ||
| * a record multiple times (e.g. a task is reassigned to the same thread | ||
| * after an EOS error). This class is idempotent to marking the same | ||
| * offsets multiple times. | ||
| */ | ||
| public class OffsetTracker { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(OffsetTracker.class); | ||
|
|
||
| private BitSet offsets = new BitSet(); | ||
| private long baseOffset; | ||
|
|
||
| public OffsetTracker(long baseOffset) { | ||
| this.baseOffset = baseOffset; | ||
| } | ||
|
|
||
| public void mark(final long offset) { | ||
| if (offset < baseOffset) { | ||
| LOG.error("Invalid offset {} lower than baseOffset {} marked", offset, baseOffset); | ||
| throw new IllegalArgumentException( | ||
|
Contributor
There was a problem hiding this comment. nit: can we log an error before throwing that also includes the value of |
||
| "Offset " + offset + " cannot be less than baseOffset " + baseOffset); | ||
| } | ||
|
|
||
| // assume that we won't be committing more than MAX_INT offsets | ||
| // in a single commit... | ||
| final int idx = Math.toIntExact(offset - baseOffset); | ||
| offsets.set(idx); | ||
| } | ||
|
|
||
| public int countAndShift(final long commitOffset) { | ||
| if (commitOffset < baseOffset) { | ||
| LOG.error("Invalid offset {} lower than baseOffset {} committed", commitOffset, baseOffset); | ||
| throw new IllegalArgumentException( | ||
| "Commit offset " + commitOffset + " cannot be less than baseOffset " + baseOffset); | ||
| } | ||
|
|
||
| final int shift = Math.toIntExact(commitOffset - baseOffset); | ||
| final int count = offsets.get(0, shift).cardinality(); | ||
|
|
||
| // shift the bitset so that commitOffset is the new base offset | ||
| // for which we track marked entries | ||
| final var old = this.offsets; | ||
agavra marked this conversation as resolved.
Outdated
Show resolved
Hide resolved
|
||
| this.offsets = new BitSet(); | ||
| for (int i = old.nextSetBit(shift); i != -1; i = old.nextSetBit(i + 1)) { | ||
| this.offsets.set(i - shift); | ||
| } | ||
|
|
||
| baseOffset = commitOffset; | ||
| return count; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /* | ||
| * Copyright 2024 Responsive Computing, Inc. | ||
| * | ||
| * This source code is licensed under the Responsive Business Source License Agreement v1.0 | ||
| * available at: | ||
| * | ||
| * https://www.responsive.dev/legal/responsive-bsl-10 | ||
| * | ||
| * This software requires a valid Commercial License Key for production use. Trial and commercial | ||
| * licenses can be obtained at https://www.responsive.dev | ||
| */ | ||
|
|
||
| package dev.responsive.kafka.internal.clients; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.Map; | ||
| import org.apache.kafka.clients.consumer.ConsumerRecords; | ||
| import org.apache.kafka.clients.consumer.OffsetAndMetadata; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
| import org.apache.kafka.common.TopicPartition; | ||
|
|
||
| public interface OriginEventRecorder | ||
| extends ResponsiveConsumer.Listener, ResponsiveProducer.Listener { | ||
|
|
||
| @Override | ||
| <K, V> ConsumerRecords<K, V> onPoll(ConsumerRecords<K, V> records); | ||
|
|
||
| @Override | ||
| <K, V> ProducerRecord<K, V> onSend(ProducerRecord<K, V> record); | ||
|
|
||
| @Override | ||
| void onConsumerCommit(Map<TopicPartition, OffsetAndMetadata> offsets); | ||
|
|
||
| @Override | ||
| void onProducerCommit(); | ||
|
|
||
| @Override | ||
| void onSendOffsetsToTransaction( | ||
| final Map<TopicPartition, OffsetAndMetadata> offsets, | ||
| final String consumerGroupId | ||
| ); | ||
|
|
||
| @Override | ||
| void onPartitionsLost(Collection<TopicPartition> partitions); | ||
|
|
||
| @Override | ||
| void onUnsubscribe(); | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
All of this License stuff was moved to
LicenseUtils