epugh commented on code in PR #4831:
URL: https://github.com/apache/solr/pull/4831#discussion_r3902931306
##########
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
Review Comment:
is this worth an assert?
##########
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:
I had claude code review, so take this with a grain of salt, but apparently
you can have a double failure? IDK -->
This `finally` block can mask the real failure. If an earlier unit in this
drain already completed (`committableOffset >= 0`) but a *later* unit's
`isComplete()` throws (propagating a failed work item), the `finally` still
calls `updateOffset(partition, committableOffset)` →
`consumer.commitSync(...)`, which isn't guarded by try/catch. Per JLS 14.20.2,
if that `commitSync` call itself throws (e.g.
`CommitFailedException`/`TimeoutException` during a rebalance), it discards the
original work-item exception propagating from the `try` block — only the
secondary commit failure reaches `KafkaCrossDcConsumer`, and the real root
cause of the failed work item is lost from the logs.
>
> Doesn't reintroduce the data-loss bug this PR fixes (the offset being
committed here is still legitimately earned), but on a double-failure it
silently erases the diagnostic trail. Might be worth catching the `commitSync`
failure here and attaching the original via `addSuppressed`, or logging the
original before it's discarded.
##########
solr/cross-dc-manager/src/java/org/apache/solr/crossdc/manager/consumer/PartitionManager.java:
##########
@@ -41,16 +39,52 @@ public class PartitionManager {
new ConcurrentHashMap<>();
private final KafkaConsumer<String, MirroredSolrRequest<?>> consumer;
- static class PartitionWork {
+ @VisibleForTesting
+ public static class PartitionWork {
Review Comment:
I think my first thought is shouldn't Kakfa be providing this capability?
I haven't done a lot with Kafka, so my knowledge is fuzzy, but I wold have
assumed that consuming this data is soemthing that Kafka client would ensure we
don't have bad side effects? Or is kafka more limited and there fofre we have
to build this logic?
--
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]
