feat: Add SplitDiscovery for use in split enumerator #14
Conversation
updated
google-cla
bot
added
the
cla: yes
| TopicPath topicPath = TopicPath.parse(proto.getTopic()); | ||
| long partitionCount = | ||
| currentSplits.stream() | ||
| .filter(s -> s.subscriptionPath().equals(subscriptionPath)) |
There was a problem hiding this comment.
I'd argue you should fail if any of the subscription paths are wrong, but this is a nit
There was a problem hiding this comment.
I'm happy with that, done
| currentSplits.stream() | ||
| .filter(s -> s.subscriptionPath().equals(subscriptionPath)) | ||
| .mapToLong(a -> a.partition().value()) | ||
| .max() |
There was a problem hiding this comment.
this is wrong. the max existing partition number != the count. partitions are 0 indexed
There was a problem hiding this comment.
Oh, good catch. Fixed this and the test
| } | ||
|
|
||
| static SingleSubscriptionSplitDiscovery fromCheckpoint( | ||
| SplitEnumeratorCheckpoint.Discovery proto, |
There was a problem hiding this comment.
can you just put the saved partition count in this proto directly?
There was a problem hiding this comment.
Unfortunately we can't. Checkpointing can be interleaved between split discovery and when the splits are actually turned over to the assigner. If we checkpoint the partition count, we might lose some splits if a checkpoint is taken after the splits are discovered but before they're given to the assigner.
...main/java/com/google/cloud/pubsublite/flink/enumerator/SingleSubscriptionSplitDiscovery.java
Outdated
Show resolved
Hide resolved
| Map<Partition, Offset> cursorMap; | ||
| try { | ||
| cursorMap = cursorClient.listPartitionCursors(subscriptionPath).get(); | ||
| } catch (ExecutionException | InterruptedException e) { |
| adminClient, cursorClient, topicPath, subscriptionPath, partitionCount); | ||
| } | ||
|
|
||
| public List<SubscriptionPartitionSplit> discoverSplits() throws ApiException { |
There was a problem hiding this comment.
do methods on this class need to be synchronized?
There was a problem hiding this comment.
I don't think so -- discover splits is the only method which modifies state and it's never called concurrently, but at the same time, there's no performance consideration or anything, so I just made them synchronized
| SubscriptionPath subscriptionPath = SubscriptionPath.parse(proto.getSubscription()); | ||
| TopicPath topicPath = TopicPath.parse(proto.getTopic()); | ||
| long partitionCount = 0; | ||
| for (SubscriptionPartitionSplit s : currentSplits) { |
There was a problem hiding this comment.
can you put them in a Set while going through this loop, and check on line 75 that every partition < partitionCount is present?
| newPartitionCount = adminClient.getTopicPartitionCount(topicPath).get(); | ||
| } catch (Throwable t) { | ||
| throw ExtractStatus.toCanonical(t).underlying; | ||
| } |
There was a problem hiding this comment.
nit. I'd just use a try{ block around the whole function
| import java.util.List; | ||
|
|
||
| interface SplitDiscovery extends AutoCloseable { | ||
| List<SubscriptionPartitionSplit> discoverSplits() throws ApiException; |
There was a problem hiding this comment.
nit. discoverNewSplits
updated
Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
Fixes #<issue_number_goes_here> ☕️