Federated Clients: implement commitSync() and committed() #129
Conversation
| } | ||
|
|
||
| try { | ||
| Thread.sleep(Math.min(deadlineTimeMs - System.currentTimeMillis(), _retryBackoffMs)); |
There was a problem hiding this comment.
Can deadlineTimeMs - System.currentTimeMillis() be negative and cause infinite sleep?
There was a problem hiding this comment.
sleep would throw for a negative
| } | ||
|
|
||
| try { | ||
| if (!countDownLatch.await(deadlineTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS)) { |
There was a problem hiding this comment.
Same question here, not sure if we need to consider negative values for await here.
| _nonexistentTopics = Collections.emptySet(); | ||
| } | ||
|
|
||
| public LocationLookupResult(/*Map<ClusterDescriptor, T> valuesByCluster, */Set<String> nonexistentTopics) { |
There was a problem hiding this comment.
should you just delete the commented out part ?
| } | ||
|
|
||
| // ATTN: UnknownTopicOrPartitionException may be received - this is a retriable exception.. | ||
| // if not resolved by the time, timeout exception |
There was a problem hiding this comment.
"by the time" - "within the timeout" ?
| long now = System.currentTimeMillis(); | ||
| long deadlineTimeMs = now + timeout.toMillis(); | ||
| while (now < deadlineTimeMs) { | ||
| PartitionKeyedMapLookupResult offsetsByClusterResult = getPartitionKeyedMapsByCluster(offsets); |
There was a problem hiding this comment.
if the while() loop executes multiple times (because some topics done exist?) it will commit the same offsets for existing topicPartitions N times. i think on successful commit the TPs committed should be taken out and not committed over again - its just wasteful
| } | ||
|
|
||
| @Override | ||
| synchronized public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) { | ||
| throw new UnsupportedOperationException("Not implemented yet"); | ||
| if (partition == null) { |
There was a problem hiding this comment.
probably want to null check timeout as well
| @@ -552,17 +589,48 @@ synchronized public long position(TopicPartition partition, Duration timeout) { | |||
|
|
|||
| @Override | |||
| synchronized public OffsetAndMetadata committed(TopicPartition partition) { | |||
| throw new UnsupportedOperationException("Not implemented yet"); | |||
|
|
|||
| return committed(partition, _defaultApiTimeout); | |||
| } | |||
|
|
|||
| @Override | |||
| synchronized public OffsetAndMetadata committed(TopicPartition partition, Duration timeout) { | |||
There was a problem hiding this comment.
does vanilla block here for non-existent TPs? if so, we should probably document this
No description provided.