-
Notifications
You must be signed in to change notification settings - Fork 13.3k
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.
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
[FLINK-9969][batch] Dispose InMemorySorters created by the UnilateralSortMerger #6479
Conversation
6a36633
to
a1a6ca0
Compare
if (initialSerializer == null) { | ||
typeSerializer = typeSerializerFactory.getSerializer(); | ||
} else { | ||
typeSerializer = initialSerializer; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
couldn't we simplify this by always discarding the serializer created in the constructor? We could then remove the initialSerializer
field and always create a new serializer here.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, the initialSerializer
is an optimization to decrease the number of created TypeSerializer
instances at the cost of a slightly more complicated create
method. I guess both ways should work given that the UnilateralSortMerger
does not create thousands of InMemorySorters
.
new DefaultInMemorySorterFactory<>(serializerFactory, comparator, THRESHOLD_FOR_IN_PLACE_SORTING)); | ||
} | ||
|
||
protected UnilateralSortMerger( |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@VisibleForTesting
?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
why? we might config which InMemorySorterFactory
to use if needed. semantically it's no need to be @VisibleForTesting
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I agree with @tisonkun
import java.util.ArrayList; | ||
import java.util.Hashtable; | ||
import java.util.Iterator; | ||
import java.util.NoSuchElementException; | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This change can be suppressed if merged.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
You're right it's unrelated. I will remove it from the commit.
a1a6ca0
to
685293d
Compare
…SortMerger This commit changes the behaviour of the UnilateralSortMerger to keep references of the created InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep references. This closes apache#6479.
To guard against memory leaks, the Task releases the reference to its AbstractInvokable when it shuts down or cancels. This closes apache#6480.
…SortMerger This commit changes the behaviour of the UnilateralSortMerger to keep references of the created InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep references. This closes apache#6479.
685293d
to
0278395
Compare
…SortMerger This commit changes the behaviour of the UnilateralSortMerger to keep references of the created InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep references. This closes apache#6479.
…SortMerger This commit changes the behaviour of the UnilateralSortMerger to keep references of the created InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep references. This closes apache#6479.
…SortMerger This commit changes the behaviour of the UnilateralSortMerger to keep references of the created InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep references. This closes #6479.
…SortMerger This commit changes the behaviour of the UnilateralSortMerger to keep references of the created InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep references. This closes #6479.
What is the purpose of the change
This commit changes the behaviour of the UnilateralSortMerger to keep references of the created
InMemorySorters in order to explicitly dispse them when the sort merger is closed. This prevents
that InMemorySorters leak and block the garbage collection of MemorySegments to which they keep
references.
cc @StephanEwen
Brief change log
InMemorySorterFactory
to makeUnilateralSortMerger
testableInMemorySorters
inUnilateralSortMerger
InMemorySorters
before releasing memory toMemoryManager
Verifying this change
UnilateralSortMergerTest#testInMemorySorterDisposal
Does this pull request potentially affect one of the following parts:
@Public(Evolving)
: (no)Documentation