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

Implement Comparable interface on StreamCut objects #7032

Open
derekm opened this issue Mar 28, 2023 · 2 comments · May be fixed by #7033
Open

Implement Comparable interface on StreamCut objects #7032

derekm opened this issue Mar 28, 2023 · 2 comments · May be fixed by #7033

Comments

@derekm
Copy link
Contributor

derekm commented Mar 28, 2023

Is your feature request related to a problem? Please describe.
When implementing a Pravega ingestion plugin for Apache Druid, Druid internals want to track ingestion offsets, which in Pravega are called StreamCuts, and these offsets need to implement the Comparable interface.

Describe the solution you'd like
public interface StreamCut extends Serializable, Comparable<StreamCut>

Describe alternatives you've considered
We've considered using EventPointers or reader Positions as the offset we let Druid track, but for various reasons having Druid track StreamCuts is preferrable. (Reasons relating to restarting new reader groups at previous offsets, composing Positions from several readers into a new StreamCut, acquiring subsequent segments after a scaling event or epoch, etc.)

Additional context
We need to wrap whatever we use as Pravega offset in a Druid OrderedSequenceNumber: https://github.com/apache/druid/blob/master/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/common/OrderedSequenceNumber.java#L34
@derekm derekm linked a pull request Mar 30, 2023 that will close this issue
Open
@RaulGracia
Copy link
Contributor

There is already a StreamCut comparison logic implemented in the Controller:

pravega/controller/src/main/java/io/pravega/controller/store/stream/PersistentStreamBase.java

Line 274 in f9ed2e1

public CompletableFuture<StreamCutComparison> compareStreamCuts(Map<Long, Long> streamcut1, Map<Long, Long> streamcut2,

Would it make sense to place this logic in the "shared" package, in the case that both client and server need to use it?

@derekm
Copy link
Contributor Author

derekm commented Apr 6, 2023

@RaulGracia -- There's another comparison routine in the client in ReaderGroupConfig, to verify that a "start" StreamCut comes before an "end" StreamCut:

pravega/client/src/main/java/io/pravega/client/stream/ReaderGroupConfig.java

Line 298 in f9ed2e1

private void verifyStartAndEndStreamCuts(final StreamCut startStreamCut, final StreamCut endStreamCut) {

The compareStreamCuts() routine you point out makes server calls in computeStreamCutSpan(). Is it possible to make an entirely offline comparison routine, as we're attempting here?

Also, the Controller's routine combines "equal" and "after." In order to implement Comparable#compareTo(), we need to distinguish between "equal" and "after."

It would be great to bring all these comparison routines closer together. Do you see a way to do that? I can help in this effort. We need a compareTo() in StreamCut to implement the Pravega plugin for Druid given the current state of its "seekable stream" indexing tasks and job supervisor code.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging a pull request may close this issue.

Issue 7032: Implement Comparable on StreamCut & StreamCut.UNBOUNDED derekm/pravega
2 participants
@derekm @RaulGracia