You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lkokhreidze (via GitHub)" <gi...@apache.org> on 2023/10/10 12:57:14 UTC

[PR] StateRestoreListener#onRestoreSuspended is never called because wrapp…er DelegatingStateRestoreListener doesn't implement onRestoreSuspended [kafka]

lkokhreidze opened a new pull request, #14519:
URL: https://github.com/apache/kafka/pull/14519

   With https://issues.apache.org/jira/browse/KAFKA-10575 `StateRestoreListener#onRestoreSuspended` was added. But local tests show that it is never called because `DelegatingStateRestoreListener` was not updated to call a new method
   
   
   ### 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

Re: [PR] KAFKA-15571 / `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1758663469

   By the way, make sure to format the PR titles exactly as recommended -- there are some integrations that regex match and rely on this (you had a slash `/`  instead of a colon `:`)


-- 
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

Re: [PR] KAFKA-15571 / `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on code in PR #14519:
URL: https://github.com/apache/kafka/pull/14519#discussion_r1354484003


##########
streams/src/test/java/org/apache/kafka/streams/integration/StateRestoreListerIntegrationTest.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.MockTime;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.processor.StateRestoreListener;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.junit.rules.Timeout;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static org.apache.kafka.streams.Topology.AutoOffsetReset.EARLIEST;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.junit.Assert.assertTrue;
+
+@Category(IntegrationTest.class)
+public class StateRestoreListerIntegrationTest {
+    private static final Duration RESTORATION_DELAY = Duration.ofSeconds(1);
+
+    @Rule
+    public Timeout globalTimeout = Timeout.seconds(600);
+
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);

Review Comment:
   Done. Moved test into the `RestoreIntegrationTest`.



-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1759069004

   Thanks Sophie. I will do that.


-- 
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

Re: [PR] KAFKA-15571 / `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1756120126

   Hi @cadonna,
   I've added an integration test. It seemed the safest way to test out the new functionality.


-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman merged PR #14519:
URL: https://github.com/apache/kafka/pull/14519


-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1772643223

   Hi @divijvaidya. Thanks for pointing that out, I will take a look. I assigned ticket to myself.


-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "divijvaidya (via GitHub)" <gi...@apache.org>.
divijvaidya commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1772557774

   Hi @lkokhreidze  - the test added in the PR `shouldInvokeUserDefinedGlobalStateRestoreListener` has been flaky since it was added. You can find the flaky build on trunk using the link https://ge.apache.org/scans/tests?search.relativeStartTime=P28D&search.rootProjectNames=kafka&search.tags=trunk&search.timeZoneId=Europe%2FBerlin&tests.container=org.apache.kafka.streams.integration.RestoreIntegrationTest&tests.test=shouldInvokeUserDefinedGlobalStateRestoreListener() 
   
   Could we please pick up a follow-up JIRA to fix the flakiness?
   
   ![Screenshot 2023-10-20 at 13 19 20](https://github.com/apache/kafka/assets/71267/c34921f8-2e1a-415e-bcba-03793438d9a6)
   
   (cc: @cadonna @ableegoldman)


-- 
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

Re: [PR] KAFKA-15571 / `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1755442563

   Thanks @cadonna I will continue working on it.


-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1759052065

   I would personally love to see a 3.5.2 release, but having managed both a bugfix release myself in the past, I can attest that it's not a trivial amount of work. Usually we don't do a bugfix release for a single broken feature, only for security patches or severe regressions
   
   That said, I actually do think we'll get a 3.5.2 release since the 3.5.1 release was rushed out the door for a CVE only like a month or so after 3.5.0 was released. Given that 3.6.0 was just released and we're starting to think about 3.7, I would expect someone to pick up 3.5.2 relatively soon
   
   Of course, even if we won't do an entire release to fix a single feature, you might be able to use this as leverage to put emphasis on getting it out sooner rather than later. I would say just go ahead and send something to the dev mailing list to inquire about the status of 3.5.2, and mention that you're waiting on it to get a fix for a feature that was supposed to be new in 3.5. 


-- 
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

Re: [PR] KAFKA-15571 / `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on code in PR #14519:
URL: https://github.com/apache/kafka/pull/14519#discussion_r1354282349


##########
streams/src/test/java/org/apache/kafka/streams/integration/StateRestoreListerIntegrationTest.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.MockTime;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.processor.StateRestoreListener;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.junit.rules.Timeout;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static org.apache.kafka.streams.Topology.AutoOffsetReset.EARLIEST;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.junit.Assert.assertTrue;
+
+@Category(IntegrationTest.class)
+public class StateRestoreListerIntegrationTest {
+    private static final Duration RESTORATION_DELAY = Duration.ofSeconds(1);
+
+    @Rule
+    public Timeout globalTimeout = Timeout.seconds(600);
+
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);

Review Comment:
   Will do, thanks.



-- 
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

Re: [PR] KAFKA-15571 / `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #14519:
URL: https://github.com/apache/kafka/pull/14519#discussion_r1354281780


##########
streams/src/test/java/org/apache/kafka/streams/integration/StateRestoreListerIntegrationTest.java:
##########
@@ -0,0 +1,302 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.integration;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.IntegerSerializer;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.util.MockTime;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
+import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
+import org.apache.kafka.streams.kstream.Consumed;
+import org.apache.kafka.streams.processor.StateRestoreListener;
+import org.apache.kafka.test.IntegrationTest;
+import org.apache.kafka.test.TestUtils;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+import org.junit.rules.Timeout;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static org.apache.kafka.streams.Topology.AutoOffsetReset.EARLIEST;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.junit.Assert.assertTrue;
+
+@Category(IntegrationTest.class)
+public class StateRestoreListerIntegrationTest {
+    private static final Duration RESTORATION_DELAY = Duration.ofSeconds(1);
+
+    @Rule
+    public Timeout globalTimeout = Timeout.seconds(600);
+
+    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3);

Review Comment:
   Whoops, forgot to add this comment before submitting the review -- anyways:
   
   Please try to avoid adding any new integration test classes that have to use an actual embedded cluster (which this one definitely does) -- it has a nontrivial setup time so there's a significant overhead for each new integration test class that has to set one up. Honestly there are probably at least a few existing integration test classes that could/should be consolidated...
   
   I think we can probably find an existing integration test file that would suit this -- probably the `RestoreIntegrationTest` makes the most sense, can you take a look at moving this over there?



-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "lkokhreidze (via GitHub)" <gi...@apache.org>.
lkokhreidze commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1758991030

   Many thanks @ableegoldman for merging the PR 🙇 
   What do you think about 3.5.2 release with a fix?
   I'm no expert with the Kafka release processes, but given the feature was implemented in 3.5 it seems logical to release a new version with a fix? But I also understand it maybe too much to release the whole Kafka project just for this.


-- 
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

Re: [PR] KAFKA-15571: `StateRestoreListener#onRestoreSuspended` is never called because `DelegatingStateRestoreListener` doesn't implement `onRestoreSuspended` [kafka]

Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on PR #14519:
URL: https://github.com/apache/kafka/pull/14519#issuecomment-1758665862

   Merged to trunk and will backport to 3.6 and 3.5


-- 
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