-
Notifications
You must be signed in to change notification settings - Fork 4
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.
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
some logging and assertions we added for antithesis tests #321
base: main
Are you sure you want to change the base?
some logging and assertions we added for antithesis tests #321
Conversation
…ests" (#…" This reverts commit 7ff8d69.
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.
some suggestions for improved logging
@@ -505,6 +544,8 @@ private int finalizeAtLeastOneEvent() { | |||
*/ | |||
private void maybeBackOffEnqueuingNewEventWithKey(final KIn key) { | |||
while (schedulingQueue.keyQueueIsFull(key)) { | |||
// TODO: switch to a periodic log | |||
log.debug("key queue is full. back off until there is room on key queue"); |
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.
nit: let's log which key is backing off, size of queue and back off time (if that's easily accessible)
@@ -110,6 +110,10 @@ public void removeProcessor(final String processorName, final int partition) { | |||
if (inFlightForTask != null) { | |||
log.info("Cancelling {} pending records for {}[{}]", | |||
inFlightForTask.size(), processorName, partition); | |||
if (!inFlightForTask.isEmpty()) { | |||
log.info("ANTITHESIS SOMETIMES: cancelling {} pending records for {}[{}]", |
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.
is ANTITHESIS SOMETIMES
a pragma?
|
||
int size(); | ||
|
||
int blockedEntries(); |
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.
nit: numBlockedEntries()
@@ -409,6 +410,18 @@ private boolean triggerFlush() { | |||
} | |||
|
|||
public void flush(final long consumedOffset) { | |||
if (consumedOffset < lastFlushedOffset) { | |||
log.error("trying to commit an offset {} older than last flushed {}", |
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.
when can this happen? Can we make this error message more actionable (or is it safe to ignore/just got fenced?)
@@ -68,6 +68,12 @@ public synchronized List<ResponsiveStoreRegistration> getRegisteredStoresForChan | |||
"there should always be a store for the thread (%s) if there are stores registered " | |||
+ "for this topic partition (%s)", threadId, topicPartition)); | |||
} | |||
if (storesForThread.size() > 1) { | |||
LOGGER.warn("found more than 1 registration: {}", |
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.
probably already in each registration but let's just add found more than 1 registration for {threadId}: {}"
No description provided.