Skip to content
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

SNOW-1759821 - [GitHub] Too many Warnings: "might have an offset token mismatch based on the provided offset token verification logic" #971

Merged
merged 5 commits into from
Oct 28, 2024
Merged

SNOW-1759821 - [GitHub] Too many Warnings: "might have an offset token mismatch based on the provided offset token verification logic" #971

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a3044d4
SNOW-1759821
sfc-gh-achyzy Oct 23, 2024
785a6ea
Fix "expected offset 0 but received 0" messages in tests
sfc-gh-achyzy Oct 26, 2024
1cada6f
Introduce new config enable.streaming.channel.offset.verification for…
sfc-gh-achyzy Oct 26, 2024
959b74e
Code formatting fix
sfc-gh-achyzy Oct 26, 2024
cc3d6d3
Config builder
sfc-gh-mbobowski Oct 28, 2024
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
8 changes: 3 additions & 5 deletions src/main/java/com/snowflake/kafka/connector/internal/streaming/StreamingUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,16 +58,14 @@ public class StreamingUtils {
public static final String STREAMING_CONSTANT_OAUTH_CLIENT_SECRET = "oauth_client_secret";
public static final String STREAMING_CONSTANT_OAUTH_REFRESH_TOKEN = "oauth_refresh_token";

// Offset verification function to verify that the current start offset has to be the previous end
// offset + 1, note that there are some false positives when SMT is used.
// Offset verification function to verify that the current start offset has to incremental,
// note that there are some false positives when SMT is used.
public static final OffsetTokenVerificationFunction offsetTokenVerificationFunction =
(prevBatchEndOffset, curBatchStartOffset, curBatchEndOffset, rowCount) -> {
if (prevBatchEndOffset != null && curBatchStartOffset != null) {
long curStart = Long.parseLong(curBatchStartOffset);
long prevEnd = Long.parseLong(prevBatchEndOffset);
if (curStart != prevEnd + 1) {
return false;
}
return curStart > prevEnd;
}
return true;
};
Expand Down
2 changes: 1 addition & 1 deletion ...test/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannelTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1283,7 +1283,7 @@ public void testOffsetTokenVerificationFunction() {
Assert.assertTrue(StreamingUtils.offsetTokenVerificationFunction.verify("1", "2", "4", 2));
Assert.assertTrue(StreamingUtils.offsetTokenVerificationFunction.verify("1", "2", null, 1));
Assert.assertTrue(StreamingUtils.offsetTokenVerificationFunction.verify(null, null, null, 0));
Assert.assertFalse(StreamingUtils.offsetTokenVerificationFunction.verify("1", "3", "4", 3));
Assert.assertTrue(StreamingUtils.offsetTokenVerificationFunction.verify("1", "3", "4", 3));
Assert.assertFalse(StreamingUtils.offsetTokenVerificationFunction.verify("2", "1", "4", 3));
}

Expand Down
Loading