You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "gharris1727 (via GitHub)" <gi...@apache.org> on 2023/03/08 23:12:37 UTC
[GitHub] [kafka] gharris1727 opened a new pull request, #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
gharris1727 opened a new pull request, #13367:
URL: https://github.com/apache/kafka/pull/13367
This fixes a regression in KAFKA-12468 and must accompany that patch in all releases.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1138946676
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationTransactionsTest.java:
##########
@@ -45,7 +45,13 @@ public void startClusters() throws Exception {
backupBrokerProps.put("transaction.state.log.min.isr", "1");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "embedded-kafka-0");
- super.startClusters();
+ startClusters(new HashMap<String, String>() {{
+ put("topics", "test-topic-.*, primary.test-topic-.*, backup.test-topic-.*");
+ put(PRIMARY_CLUSTER_ALIAS + "->" + BACKUP_CLUSTER_ALIAS + ".enabled", "true");
+ put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS + ".enabled", "true");
+ // This is not necessary for this test, but is not tested elsewhere.
+ put("offset.lag.max", "0");
Review Comment:
Don't we assume that the value of this property is non-zero in `MirrorConnectorsIntegrationBaseTest::waitForConsumerGroupFullSync`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1134035661
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java:
##########
@@ -316,13 +315,14 @@ static class PartitionState {
// true if we should emit an offset sync
boolean update(long upstreamOffset, long downstreamOffset) {
- long upstreamStep = upstreamOffset - lastSyncUpstreamOffset;
- long downstreamTargetOffset = lastSyncDownstreamOffset + upstreamStep;
+ // This value is what OffsetSyncStore::translateOffsets would compute for this offset given the last sync.
+ // Because this method is called at most once for each upstream offset, simplify upstreamStep to 1.
+ // TODO: share common implementation to enforce this relationship
+ long downstreamTargetOffset = lastSyncDownstreamOffset + 1;
Review Comment:
I've started to like the name `downstreamTargetOffset` less and less, and the need for 2-3 lines of comments to explain what it's supposed to represent might hint that a better name could help more here.
We can remove this variable entirely and just inline the value `lastSyncDownstreamOffset + 1` in its place, with a comment explaining the `+ 1` part. Or we could rename to something like `translatedLastDownstreamOffset`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1135661761
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java:
##########
@@ -315,18 +314,16 @@ static class PartitionState {
// true if we should emit an offset sync
boolean update(long upstreamOffset, long downstreamOffset) {
- // This value is what OffsetSyncStore::translateOffsets would compute for this offset given the last sync.
- // Because this method is called at most once for each upstream offset, simplify upstreamStep to 1.
+ // Emit an offset sync if any of the following conditions are true
+ boolean noPreviousSyncThisLifetime = lastSyncDownstreamOffset == -1L;
+ // the OffsetSync::translateDownstream method will translate this offset 1 past the last sync, so add 1.
// TODO: share common implementation to enforce this relationship
- long downstreamTargetOffset = lastSyncDownstreamOffset + 1;
- if (lastSyncDownstreamOffset == -1L
- || downstreamOffset - downstreamTargetOffset >= maxOffsetLag
- || upstreamOffset - previousUpstreamOffset != 1L
Review Comment:
Why are we removing this condition?
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java:
##########
@@ -315,18 +314,16 @@ static class PartitionState {
// true if we should emit an offset sync
boolean update(long upstreamOffset, long downstreamOffset) {
- // This value is what OffsetSyncStore::translateOffsets would compute for this offset given the last sync.
- // Because this method is called at most once for each upstream offset, simplify upstreamStep to 1.
+ // Emit an offset sync if any of the following conditions are true
+ boolean noPreviousSyncThisLifetime = lastSyncDownstreamOffset == -1L;
+ // the OffsetSync::translateDownstream method will translate this offset 1 past the last sync, so add 1.
// TODO: share common implementation to enforce this relationship
- long downstreamTargetOffset = lastSyncDownstreamOffset + 1;
- if (lastSyncDownstreamOffset == -1L
- || downstreamOffset - downstreamTargetOffset >= maxOffsetLag
- || upstreamOffset - previousUpstreamOffset != 1L
Review Comment:
Why are we removing this condition? Is that necessary for this fix?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] gharris1727 commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1135890001
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java:
##########
@@ -315,18 +314,16 @@ static class PartitionState {
// true if we should emit an offset sync
boolean update(long upstreamOffset, long downstreamOffset) {
- // This value is what OffsetSyncStore::translateOffsets would compute for this offset given the last sync.
- // Because this method is called at most once for each upstream offset, simplify upstreamStep to 1.
+ // Emit an offset sync if any of the following conditions are true
+ boolean noPreviousSyncThisLifetime = lastSyncDownstreamOffset == -1L;
+ // the OffsetSync::translateDownstream method will translate this offset 1 past the last sync, so add 1.
// TODO: share common implementation to enforce this relationship
- long downstreamTargetOffset = lastSyncDownstreamOffset + 1;
- if (lastSyncDownstreamOffset == -1L
- || downstreamOffset - downstreamTargetOffset >= maxOffsetLag
- || upstreamOffset - previousUpstreamOffset != 1L
Review Comment:
Oops this leaked from another fix, reverting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1138945859
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationTransactionsTest.java:
##########
@@ -45,7 +45,13 @@ public void startClusters() throws Exception {
backupBrokerProps.put("transaction.state.log.min.isr", "1");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "embedded-kafka-0");
- super.startClusters();
+ startClusters(new HashMap<String, String>() {{
+ put("topics", "test-topic-.*, primary.test-topic-.*, backup.test-topic-.*");
+ put(PRIMARY_CLUSTER_ALIAS + "->" + BACKUP_CLUSTER_ALIAS + ".enabled", "true");
+ put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS + ".enabled", "true");
+ // This is not necessary for this test, but is not tested elsewhere.
+ put("offset.lag.max", "0");
Review Comment:
Don't we assume that the value of this property is non-zero in `MirrorConnectorsIntegrationBaseTest::waitForConsumerGroupFullSync`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante merged pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante merged PR #13367:
URL: https://github.com/apache/kafka/pull/13367
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1134001434
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskTest.java:
##########
@@ -99,11 +99,13 @@ public void testOffsetSync() {
partitionState.reset();
assertFalse(partitionState.update(3, 152), "no sync");
partitionState.reset();
- assertFalse(partitionState.update(4, 153), "no sync");
+ assertTrue(partitionState.update(4, 153), "one past target offset");
partitionState.reset();
assertFalse(partitionState.update(5, 154), "no sync");
partitionState.reset();
- assertTrue(partitionState.update(6, 205), "one past target offset");
+ assertFalse(partitionState.update(6, 203), "one past target offset");
Review Comment:
Copy-paste error?
```suggestion
assertFalse(partitionState.update(6, 203), "no sync");
```
##########
connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java:
##########
@@ -316,13 +315,14 @@ static class PartitionState {
// true if we should emit an offset sync
boolean update(long upstreamOffset, long downstreamOffset) {
- long upstreamStep = upstreamOffset - lastSyncUpstreamOffset;
- long downstreamTargetOffset = lastSyncDownstreamOffset + upstreamStep;
+ // This value is what OffsetSyncStore::translateOffsets would compute for this offset given the last sync.
+ // Because this method is called at most once for each upstream offset, simplify upstreamStep to 1.
Review Comment:
This comment mentions `upstreamStep` but we no longer have a field named that.
Is this line really necessary? Do you think it would be clear enough with just the comments on lines 318 and 320?
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java:
##########
@@ -870,6 +864,20 @@ protected static void assertMonotonicCheckpoints(EmbeddedConnectCluster cluster,
}
}
+ protected static void assertDownstreamRedeliveriesBoundedByMaxLag(Consumer<byte[], byte[]> targetConsumer) {
+ ConsumerRecords<byte[], byte[]> records = targetConsumer.poll(CONSUMER_POLL_TIMEOUT_MS);
+ // After a full sync, there should be at most offset.lag.max records per partition consumed by both upstream and downstream consumers.
+ Map<TopicPartition, Integer> perPartitionCount = new HashMap<>();
+ for (ConsumerRecord<byte[], byte[]> record : records) {
+ TopicPartition tp = new TopicPartition(record.topic(), record.partition());
+ int previous = perPartitionCount.getOrDefault(tp, 0);
+ perPartitionCount.put(tp, previous + 1);
+ }
+ for (Map.Entry<TopicPartition, Integer> entry : perPartitionCount.entrySet()) {
+ assertTrue(entry.getValue() < OFFSET_LAG_MAX, "downstream consumer is re-reading more than " + OFFSET_LAG_MAX + " records from" + entry.getKey());
+ }
Review Comment:
This can be simplified:
```suggestion
for (TopicPartition tp : records.partitions()) {
int count = records.records(tp).size();
assertTrue(count < OFFSET_LAG_MAX, "downstream consumer is re-reading more than " + OFFSET_LAG_MAX + " records from" + tp);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] gharris1727 commented on pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #13367:
URL: https://github.com/apache/kafka/pull/13367#issuecomment-1468475024
> One high-level thought: it seems like we've elected to drop integration testing coverage for offset.lag.max = 0 and replace it with offset.lag.max = 10. Do you think there's any benefit in retaining at least one case where the max lag is 0?
Before KAFKA-12468, we did not have offset.lag.max=0 coverage, so this is returning to the coverage we had before that patch. Since offset.lag.max=0 does have some effect, i'll leave it active in the transactional test. I didn't think it was valuable enough to justify a new test suite's runtime, but using it in an existing test shouldn't be too harmful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] gharris1727 commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1139409839
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationTransactionsTest.java:
##########
@@ -45,7 +45,13 @@ public void startClusters() throws Exception {
backupBrokerProps.put("transaction.state.log.min.isr", "1");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "embedded-kafka-0");
- super.startClusters();
+ startClusters(new HashMap<String, String>() {{
+ put("topics", "test-topic-.*, primary.test-topic-.*, backup.test-topic-.*");
+ put(PRIMARY_CLUSTER_ALIAS + "->" + BACKUP_CLUSTER_ALIAS + ".enabled", "true");
+ put(BACKUP_CLUSTER_ALIAS + "->" + PRIMARY_CLUSTER_ALIAS + ".enabled", "true");
+ // This is not necessary for this test, but is not tested elsewhere.
+ put("offset.lag.max", "0");
Review Comment:
The way the assertions were written, they would still be correct for offset.lag.max=0, just weaker than necessary.
In the new parameterization, if we run with offset.lag.max=0, the assertions will also get stronger to match.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] gharris1727 commented on pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "gharris1727 (via GitHub)" <gi...@apache.org>.
gharris1727 commented on PR #13367:
URL: https://github.com/apache/kafka/pull/13367#issuecomment-1472825496
@C0urante I refactored it so that we have zero and nonzero test cases in each suite. There's a little bit of duplication around the IdentityReplicationIntegrationTest that i'll address in a follow-up PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #13367:
URL: https://github.com/apache/kafka/pull/13367#issuecomment-1477875999
Merged and backported to 3.3 and 3.4
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] C0urante commented on a diff in pull request #13367: KAFKA-14797: Emit offset sync when offset translation lag would exceed max.offset.lag
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #13367:
URL: https://github.com/apache/kafka/pull/13367#discussion_r1142683836
##########
connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationBaseTest.java:
##########
@@ -416,6 +417,15 @@ public void testReplicationWithEmptyPartition() throws Exception {
@Test
public void testOneWayReplicationWithAutoOffsetSync() throws InterruptedException {
+ testOneWayReplicationWithOffsetSyncs(OFFSET_LAG_MAX);
+ }
+
+ @Test
+ public void testOneWayReplicationWithFrequentOffsetSyncs() throws InterruptedException {
+ testOneWayReplicationWithOffsetSyncs(0);
+ }
+
+ public void testOneWayReplicationWithOffsetSyncs(int offsetLagMax) throws InterruptedException {
Review Comment:
Nit: this may make more sense as `protected`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org