sigram commented on code in PR #4831:
URL: https://github.com/apache/solr/pull/4831#discussion_r3912235117
##########
solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/PartitionManager.java:
##########
@@ -73,50 +107,74 @@ public PartitionWork getPartitionWork(TopicPartition
partition) {
});
}
- public void checkOffsetUpdates() throws Throwable {
+ public void checkOffsetsAndUpdate() throws Throwable {
for (TopicPartition partition : partitionWorkMap.keySet()) {
- checkForOffsetUpdates(partition);
+ checkOffsetsAndUpdate(partition);
}
}
- void checkForOffsetUpdates(TopicPartition partition) throws Throwable {
- synchronized (partition) {
- PartitionWork work;
- if ((work = partitionWorkMap.get(partition)) != null) {
- WorkUnit workUnit = work.partitionQueue.peek();
- if (workUnit != null) {
- boolean allFuturesDone = true;
- for (Future<?> future : workUnit.workItems) {
- if (!future.isDone()) {
- if (log.isTraceEnabled()) {
- log.trace("Future for update is not done topic={}",
partition.topic());
- }
- allFuturesDone = false;
- break;
- }
-
- try {
- future.get();
- } catch (InterruptedException e) {
- log.error("Error updating offset for partition: {}", partition,
e);
- throw e;
- } catch (ExecutionException e) {
- log.error("Error updating offset for partition: {}", partition,
e);
- throw e.getCause();
- }
-
- if (log.isTraceEnabled()) {
- log.trace("Future for update is done topic={}",
partition.topic());
- }
+ void checkOffsetsAndUpdate(TopicPartition partition) throws Throwable {
+ // can't synchronize on the argument (equal but distinct object for
different threads)
+ // sync on the PartitionWork instead, which is unique per partition and
shared by all threads
+ // that work on that partition.
+ final PartitionWork partitionWork = partitionWorkMap.get(partition);
+ if (partitionWork == null) {
+ // normally impossible because consumer should always call
#getPartitionWork first
+ // which creates the instance if it doesn't exist.
+ return;
+ }
+ synchronized (partitionWork) {
+ // remove every completed work unit at the head of the queue, stopping
at the first one
+ // that is still in flight - a work unit's offset may only be committed
once all of the
+ // work units before it have been committed too.
+ long committableOffset = -1;
+ WorkUnit workUnit;
+ try {
+ while ((workUnit = partitionWork.partitionQueue.peek()) != null) {
+ if (!isComplete(workUnit, partition)) {
+ break;
}
+ // remove completed unit
+ partitionWork.partitionQueue.poll();
+ committableOffset = workUnit.nextOffset;
+ }
+ } finally {
+ // commit whatever progress was already verified in this drain, even
if a later
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
