You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "RainerMatthiasS (via GitHub)" <gi...@apache.org> on 2023/08/18 11:40:05 UTC
[GitHub] [flink] RainerMatthiasS opened a new pull request, #23239: [FLINK-26585] replace implementation of MultiStateKeyIterator with Stream-free implementation
RainerMatthiasS opened a new pull request, #23239:
URL: https://github.com/apache/flink/pull/23239
## Contribution Checklist
- `Pending:` Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
## What is the purpose of the change
Avoids a known flaw in Stream.flatMap, see https://bugs.openjdk.org/browse/JDK-8267359. Avoids potential OOM Error for
keyed state with many keys when reading them with state processor api.
## Brief change log
- replaced the original implementation of MultiStateKeyIterator with a Java-Stream-free implementation.
- this fixes a known flaw in Stream.flatMap which leads to complete enumeration of inner iterator
and buffering of all elements with potential for a OOM error for state with many keys
- manual implementation of nested iterator avoids this problem
## Verifying this change
This change adds a tests MultiStateKeyIteratorTest#testIteratorPullsSingleKeyFromAllDescriptors that fails with the previous implementation and succeeds with the new implementation.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- new code should be marginally faster (not verified)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? JavaDocs
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #23239: [FLINK-26585] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1683800450
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1403e3368cb19f089e3b67511c3db2d008d87b80",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1403e3368cb19f089e3b67511c3db2d008d87b80",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1403e3368cb19f089e3b67511c3db2d008d87b80 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1306984228
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
}
@Override
public K next() {
- currentKey = internal.next();
- return currentKey;
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ } else {
+ currentKey = this.innerIter.next();
+ return currentKey;
+ }
}
- /** Removes the current key from <b>ALL</b> known states in the state backend. */
Review Comment:
It is an oversight ...
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1318144505
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +226,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
@masteryhx Is there anything else I can do in order to progress this 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1740768001
Hi @masteryhx , thank you for the review again, I also synced my fork with the upstream, if that was what you meant by 'Could you also rebase all your commits to the master ?'
BTW: removing `boolean hasnext = ` created an additional warning on build, that's why I added it in the first place.
Sincere greeting
Thias
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1794343153
> Hi @masteryhx , thank you for the review again, I also synced my fork with the upstream, if that was what you meant by 'Could you also rebase all your commits to the master ?' BTW: removing `boolean hasnext = ` created an additional warning on build, that's why I added it in the first place.
>
> Sincere greeting
>
> Thias
Sorry for the late reply due to my personal business.
Thanks for the update. Current PR LGTM.
As you could see, there are 6 'Commits'.
You could use `git rebase` to the original master and then squash them into one commit.
Then after the CI passed, I will merge 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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1306978316
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
Review Comment:
You are right, thanks. In case of state sets the inner iterator always has at least one element otherwise it wouldn't be in the outer enumeration. How ever this assumption would be a tight coupling to external components. Should I rephrase the section ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1798418690
> > > > Hi @masteryhx , thank you for the review again, I also synced my fork with the upstream, if that was what you meant by 'Could you also rebase all your commits to the master ?' BTW: removing `boolean hasnext = ` created an additional warning on build, that's why I added it in the first place.
> > > > Sincere greeting
> > > > Thias
> > >
> > >
> > > Sorry for the late reply due to my personal business. Thanks for the update. Current PR LGTM. As you could see, there are 6 'Commits'. You could use `git rebase` to the original master and then squash them into one commit. Then after the CI passed, I will merge it.
> >
> >
> > Hi @masteryhx ,
> > Sorry to bother you, I know how to rebase on my local git, the confusion is with pushing that onto my fork on Github ... If I would do that, I believe, the PR would be force-closed because of the relocated branch. I experienced that a couple of times on our local GitHub Enterprise ... Have you got any more specific instructions on how to avoid this?
> > Thias
>
> You could try to:
>
> 1. use 'git pull master --rebase' to rebase your local repo
> 2. use 'git push ' to push/force-push to your remote repo then github will update automatically
Done, I hope that did it, thanks @masteryhx again for the instructions and help
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1306983503
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
}
@Override
public K next() {
- currentKey = internal.next();
- return currentKey;
+ if (!this.hasNext()) {
Review Comment:
`this.hasNext()` is the only place that initializes the `innerIter` and therefore is neccessary for the case the caller hasn't called `hasNext()` before.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1339585015
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,13 +31,17 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: Avoids using Stream#flatMap due to a known flaw, see FLINK-26585 for more details.
Review Comment:
nit: Move the comment to L53 ?
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +226,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
+ CountingKeysKeyedStateBackend keyedStateBackend =
+ createCountingKeysKeyedStateBackend(100_000_000);
+ MultiStateKeyIterator<Integer> testedIterator =
+ new MultiStateKeyIterator<>(descriptors, keyedStateBackend);
+
+ boolean hasnext = testedIterator.hasNext();
Review Comment:
```suggestion
testedIterator.hasNext();
```
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +226,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
Okay, Thanks for the clarification.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1797370868
> > > Hi @masteryhx , thank you for the review again, I also synced my fork with the upstream, if that was what you meant by 'Could you also rebase all your commits to the master ?' BTW: removing `boolean hasnext = ` created an additional warning on build, that's why I added it in the first place.
> > > Sincere greeting
> > > Thias
> >
> >
> > Sorry for the late reply due to my personal business. Thanks for the update. Current PR LGTM. As you could see, there are 6 'Commits'. You could use `git rebase` to the original master and then squash them into one commit. Then after the CI passed, I will merge it.
>
> Hi @masteryhx ,
>
> Sorry to bother you, I know how to rebase on my local git, the confusion is with pushing that onto my fork on Github ... If I would do that, I believe, the PR would be force-closed because of the relocated branch. I experienced that a couple of times on our local GitHub Enterprise ... Have you got any more specific instructions on how to avoid this?
>
> Thias
You could try to:
1. use 'git pull <flink-repo> master --rebase' to rebase your local repo
2. use 'git push <your-repo> <your-branch>' to push/force-push to your remote repo then github will update automatically
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1385899714
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -46,47 +48,54 @@ public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ /** Avoids using Stream#flatMap due to a known flaw, see FLINK-26585 for more details. */
+ private final Iterator<? extends StateDescriptor<?, ?>> outerIter;
+
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
Review Comment:
Move L63-L64 to L68 ?
It's better to call the method after `checkNotNull`
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1386251484
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -46,47 +48,54 @@ public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ /** Avoids using Stream#flatMap due to a known flaw, see FLINK-26585 for more details. */
+ private final Iterator<? extends StateDescriptor<?, ?>> outerIter;
+
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
Review Comment:
Hi @masteryhx , done, rebased, squashed, force pushed as before :)
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1306991331
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
Review Comment:
You may missed what I mean.
I mean that why the `final` is removed.
We should avoid unnecessary modification.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1307015491
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
Review Comment:
Yeah, I think you should.
IIUC, just in this case, it may also bring some exceptions:
1. `StateDescriptor` is passed by users, which may not be correct
2. if user just registered state but not put any states, the state will be empty (only state meta is stored in the checkpoint, this also works)
Of course, even if it works till now, we also should guarantee that this class could work not coupling to other calsses.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1306992807
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -111,9 +128,4 @@ public void remove() {
}
}
}
-
- @Override
- public void close() throws Exception {
Review Comment:
Sure :smile: . Only that both implementations are unrelated and only happen to have the same name.
(Should I move this down?)
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
Review Comment:
:+1:
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +226,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
@masteryhx Is there anything else I can do in order to progress this PR?
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +226,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
No, it actually tests the 'economy' of `hasNext()` and tests that a single call to `hasNext()` does **not** iterate the whole collection (which is the case for the original implementation by means of java streams (the [JDK-8267359](https://bugs.openjdk.org/browse/JDK-8267359) flaw)).
I.e. this test is exactly as intended.
To test completeness of enumeration is done in `testIteratorPullsKeyFromAllDescriptors()`.
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +185,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
The original implementation of `MultiStateKeyIterator` would iterate all 1e8 keys and store them in a `SpinedBuffer`, just for a single initial call to `hasNext()`. The test asserts that only a single key is touched. Technically, enumerating all keys is still a correct implementation but a wasteful one that might even lead to OOM errors.
We only count how many keys are actually enumerated instead of observing the actual memory (GC) consumption, which would be too brittle as a unit test.
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +185,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
+ CountingManyKeysKeyedStateBackend keyedStateBackend =
+ createManyKeysKeyedStateBackend(100_000_000);
+ MultiStateKeyIterator<Integer> testedIterator =
+ new MultiStateKeyIterator<>(descriptors, keyedStateBackend);
+
+ boolean hasnext = testedIterator.hasNext();
+
+ Assert.assertEquals(
+ "Unexpected number of keys enumerated",
+ 1,
+ keyedStateBackend.numberOfKeysEnumerated);
+ }
+
+ /**
+ * Mockup {@link AbstractKeyedStateBackend} that counts how many keys are enumerated.
+ *
+ * <p>Generates a configured nmber of integer keys, only method actually implemented is {@link
+ * CountingManyKeysKeyedStateBackend#getKeys(java.lang.String, java.lang.Object)}
+ */
+ static class CountingManyKeysKeyedStateBackend extends AbstractKeyedStateBackend<Integer> {
Review Comment:
:+1:
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1695199973
I'll update the PR accordingly in the next days. Thanks for the review
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1311067683
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -46,47 +55,59 @@ public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
}
@Override
public boolean hasNext() {
Review Comment:
How about:
```
@Override
public boolean hasNext() {
while (innerIter == null || !innerIter.hasNext()) {
if (!outerIter.hasNext()) {
return false;
}
StateDescriptor<?, ?> descriptor = outerIter.next();
Stream<K> stream =
backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
innerIter = stream.iterator();
try {
registry.registerCloseable(stream::close);
} catch (IOException e) {
throw new RuntimeException(
"Failed to read keys from configured StateBackend", e);
}
}
return true;
}
```
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +226,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
IIUC, you want to use this case to verify the key numer you iterator is correct ?
So you should iterator until it doesn't hasNext, right ?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,13 +31,22 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
Review Comment:
IMO, this comment about why we update the logic is not necessary, which could be found in the Jira Ticket.
Or you could just add simple description before outerIter, e.g. : "Avoid using Stream#flatMap due to xxx, see FLINK-26585 for more details"
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -46,47 +55,59 @@ public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
Review Comment:
Could be "final" ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] masteryhx commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1305077917
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
Review Comment:
IIUC, this is incorrect.
After first if, innerIter maybe an empty iterator (due to lacking related states or empty states) but we could not say it doesn't hasNext (outerIter may still hasNext)
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
Review Comment:
Why the change is needed ?
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +185,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
Review Comment:
I may missed something.
This UT may still not complete, right ?
Could you explain what we are verified from L197 - L202 ?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -111,9 +128,4 @@ public void remove() {
}
}
}
-
- @Override
- public void close() throws Exception {
Review Comment:
We could avoid moving codes.
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
}
@Override
public K next() {
- currentKey = internal.next();
- return currentKey;
+ if (!this.hasNext()) {
Review Comment:
It's not necessary due to contract of `Iterator`.
This will increase unnecessary cost when using iterator.
##########
flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/MultiStateKeyIteratorTest.java:
##########
@@ -125,4 +185,117 @@ public void testIteratorRemovesFromAllDescriptors() throws Exception {
.count());
}
}
+
+ /** Test for lazy enumeration of inner iterators. */
+ @Test
+ public void testIteratorPullsSingleKeyFromAllDescriptors() throws AssertionError {
+ CountingManyKeysKeyedStateBackend keyedStateBackend =
+ createManyKeysKeyedStateBackend(100_000_000);
+ MultiStateKeyIterator<Integer> testedIterator =
+ new MultiStateKeyIterator<>(descriptors, keyedStateBackend);
+
+ boolean hasnext = testedIterator.hasNext();
+
+ Assert.assertEquals(
+ "Unexpected number of keys enumerated",
+ 1,
+ keyedStateBackend.numberOfKeysEnumerated);
+ }
+
+ /**
+ * Mockup {@link AbstractKeyedStateBackend} that counts how many keys are enumerated.
+ *
+ * <p>Generates a configured nmber of integer keys, only method actually implemented is {@link
+ * CountingManyKeysKeyedStateBackend#getKeys(java.lang.String, java.lang.Object)}
+ */
+ static class CountingManyKeysKeyedStateBackend extends AbstractKeyedStateBackend<Integer> {
Review Comment:
How about renaming to `CountingKeyNumberKeyedStateBackend` or just `CountingKeysKeyedStateBackend` ?
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
+public class MultiStateKeyIterator<K> implements CloseableIterator<K> {
private final List<? extends StateDescriptor<?, ?>> descriptors;
private final KeyedStateBackend<K> backend;
- private final Iterator<K> internal;
+ private Iterator<? extends StateDescriptor<?, ?>> outerIter;
+ private Iterator<K> innerIter;
private final CloseableRegistry registry;
private K currentKey;
public MultiStateKeyIterator(
List<? extends StateDescriptor<?, ?>> descriptors, KeyedStateBackend<K> backend) {
+
+ outerIter = descriptors.iterator();
+ innerIter = null;
+
this.descriptors = Preconditions.checkNotNull(descriptors);
this.backend = Preconditions.checkNotNull(backend);
this.registry = new CloseableRegistry();
- this.internal =
- descriptors.stream()
- .map(
- descriptor ->
- backend.getKeys(
- descriptor.getName(), VoidNamespace.INSTANCE))
- .peek(
- stream -> {
- try {
- registry.registerCloseable(stream::close);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to read keys from configured StateBackend",
- e);
- }
- })
- .flatMap(stream -> stream)
- .iterator();
+ }
+
+ @Override
+ public void close() throws Exception {
+ registry.close();
}
@Override
public boolean hasNext() {
- return internal.hasNext();
+ if ((innerIter == null || !innerIter.hasNext()) && outerIter.hasNext()) {
+ StateDescriptor<?, ?> descriptor = outerIter.next();
+ Stream<K> stream = backend.getKeys(descriptor.getName(), VoidNamespace.INSTANCE);
+ innerIter = stream.iterator();
+ try {
+ registry.registerCloseable(stream::close);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to read keys from configured StateBackend", e);
+ }
+ }
+ if (innerIter == null) {
+ return false;
+ }
+ return innerIter.hasNext();
}
@Override
public K next() {
- currentKey = internal.next();
- return currentKey;
+ if (!this.hasNext()) {
+ throw new NoSuchElementException();
+ } else {
+ currentKey = this.innerIter.next();
+ return currentKey;
+ }
}
- /** Removes the current key from <b>ALL</b> known states in the state backend. */
Review Comment:
Why removing this comment ?
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on a diff in pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on code in PR #23239:
URL: https://github.com/apache/flink/pull/23239#discussion_r1306963160
##########
flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/MultiStateKeyIterator.java:
##########
@@ -31,65 +31,82 @@
import java.io.IOException;
import java.util.Iterator;
import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.stream.Stream;
/**
* An iterator for reading all keys in a state backend across multiple partitioned states.
*
* <p>To read unique keys across all partitioned states callers must invoke {@link
* MultiStateKeyIterator#remove}.
*
+ * <p>Note: This is a replacement of the original implementation which used streams with a known
+ * flaw in the {@link Stream#flatMap(java.util.function.Function)} implementation that lead to
+ * completely enumerating and buffering nested iterators event for a single call to {@link
+ * MultiStateKeyIterator#hasNext}.
+ *
+ * @see <a
+ * href="https://bugs.openjdk.org/browse/JDK-8267359">https://bugs.openjdk.org/browse/JDK-8267359</a>
* @param <K> Type of the key by which state is keyed.
*/
@Internal
-public final class MultiStateKeyIterator<K> implements CloseableIterator<K> {
Review Comment:
Good morning
[JDK-8267359](https://bugs.openjdk.org/browse/JDK-8267359) leads to inadequate/unnecessary use of computer memory up to OOM errors when reading state sets with many keys ... the manual implementation of the nested iterator prevents this known java streams problem.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] RainerMatthiasS commented on pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1695452766
@masteryhx : thanks for the first review again, I made the requested changes, added an extra test for the `hasNext()` changes requested.
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx closed pull request #23239: [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation
URL: https://github.com/apache/flink/pull/23239
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-26585][state-processor-api] replace implementation of MultiStateKeyIterator with Stream-free implementation [flink]
Posted by "RainerMatthiasS (via GitHub)" <gi...@apache.org>.
RainerMatthiasS commented on PR #23239:
URL: https://github.com/apache/flink/pull/23239#issuecomment-1794694360
> > Hi @masteryhx , thank you for the review again, I also synced my fork with the upstream, if that was what you meant by 'Could you also rebase all your commits to the master ?' BTW: removing `boolean hasnext = ` created an additional warning on build, that's why I added it in the first place.
> > Sincere greeting
> > Thias
>
> Sorry for the late reply due to my personal business. Thanks for the update. Current PR LGTM. As you could see, there are 6 'Commits'. You could use `git rebase` to the original master and then squash them into one commit. Then after the CI passed, I will merge it.
Hi @masteryhx ,
Sorry to bother you, I know how to rebase on my local git, the confusion is with pushing that onto my fork on Github ... If I would do that, I believe, the PR would be force-closed because of the relocated branch. I experienced that a couple of times on our local GitHub Enterprise ... Have you got any more specific instructions on how to avoid this?
Thias
--
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: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org