Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

=str Tweak the stream mapAsyncPartitioned operator #676

Merged
merged 1 commit into from
Oct 9, 2023
Merged

=str Tweak the stream mapAsyncPartitioned operator #676

merged 1 commit into from
Oct 9, 2023

Conversation

He-Pin
Copy link
Member

@He-Pin He-Pin commented Sep 23, 2023 •
edited
Loading

Motivation:

  1. Tweak the mapAsyncPartitioned & mapAsyncPartitionedUnordered document
  2. Removed some methods which is used only once.
  3. Add these method to SubSource and SubFlow.

refs: #561

@He-Pin He-Pin marked this pull request as ready for review September 23, 2023 10:53
@He-Pin He-Pin force-pushed the tweak branch 2 times, most recently from 21a80c7 to ba60197 Compare September 23, 2023 11:13
@He-Pin He-Pin requested a review from pjfanning September 23, 2023 11:43
@pjfanning
Copy link
Contributor

Can you update the .md files for these 2 operations? If we are calling the partition function partitioner then that change should be in the .md files too.

@pjfanning
Copy link
Contributor

@jaceksokol could you have a look? The changes generally seem reasonable to me. I intend to do a more thorough check within the next few days.

pjfanning
pjfanning reviewed Sep 27, 2023
View reviewed changes
stream/src/main/scala/org/apache/pekko/stream/MapAsyncPartitioned.scala
// To support both fail-fast when the supervision directive is Stop
// and not calling the decider multiple times (#23888) we need to cache the decider result and re-use that
private var cachedSupervisionDirective: Option[Supervision.Directive] = None
private final class Holder[In, Out](val in: In, var out: Try[Out], val cb: AsyncCallback[Holder[In, Out]]) extends (
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could we use the name callback instead of cb?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

  • I changed it to cb as the same in Holder in MapAsync.
  • the cb is ok, Cats-Effect use cb in unsafeRunAsync too, maybe that's fine to keep it?

@jaceksokol
Copy link
Contributor

@jaceksokol could you have a look? The changes generally seem reasonable to me. I intend to do a more thorough check within the next few days.

Most of the changes are related to code style / structure - LGTM.

@He-Pin
Copy link
Member Author

He-Pin commented Sep 27, 2023

@pjfanning It is partitioner in the mapAsyncPartitioned.md

@He-Pin
Copy link
Member Author

He-Pin commented Sep 27, 2023 •
edited
Loading

@jaceksokol Thanks, I also added a description about the behavior, which the max parallelism of per partition is 1, can you confirm that too:)

@He-Pin He-Pin requested review from mdedetrich and pjfanning October 9, 2023 07:53
pjfanning
pjfanning approved these changes Oct 9, 2023
View reviewed changes
Copy link
Contributor

@pjfanning pjfanning left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

lgtm

@He-Pin He-Pin merged commit 1b1f572 into apache:main Oct 9, 2023
17 checks passed
@He-Pin He-Pin deleted the tweak branch October 9, 2023 18:34
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@pjfanning pjfanning pjfanning approved these changes

@mdedetrich mdedetrich Awaiting requested review from mdedetrich

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
4 participants
@He-Pin @pjfanning @jaceksokol @mdedetrich