Tao could you please update your patch so that it doesn't change the order of imports in BlockManager.java.
You might also want to update testMetaSaveWithOrphanedBlocks() - name it something else, because it should actually test there are no orphaned blocks.

Konstantin Shvachko
added a comment - 06/Jun/13 19:57 Tao could you please update your patch so that it doesn't change the order of imports in BlockManager.java.
You might also want to update testMetaSaveWithOrphanedBlocks() - name it something else, because it should actually test there are no orphaned blocks.

The only place where the replicationIndex is used is for this segment in computeUnderReplicatedBlocks(int).

Integer replIndex = priorityToReplIdx.get(priority);
// skip to the first unprocessed block, which is at replIndex
for (int i = 0; i < replIndex && neededReplicationsIterator.hasNext(); i++) {
neededReplicationsIterator.next();
}

Because our condition includes .hasNext() we will stop before trying to go outside our iterator. So we are safe with this change.

The reason we cannot decrementReplicationIndex(), or why it would be difficult to, is because we cannot guarantee that we will have actually removed the block from any SPECIFIC priority from within neededReplications. We do not know what the priority level of this block is within neededReplications when we try to do BlockManager.removeBlock(Block). If you look at remove(Block, int) in UnderReplicatedBlocks, it will try to remove it from the first queue, and if that fails, then tries to remove from all the queues. Therefore we will not be able to determine WHICH replicationIndex to decrement appropriately. However, as noted above, we are safe-guarded by the iterator nonetheless.

Now with that being said, I can understand from a consistency point as to why we would want to decrement the replicationIndex. However, in order to do it, we will need to add a method in UnderReplicatedBlocks such that we attempt to remove from a priority, and return true ONLY if we actually removed it from that priority; then we can safely decrement the replicationIndex of that priority.

Plamen Jeliazkov
added a comment - 07/Jun/13 19:36 The only place where the replicationIndex is used is for this segment in computeUnderReplicatedBlocks(int).
Integer replIndex = priorityToReplIdx.get(priority);
// skip to the first unprocessed block, which is at replIndex
for ( int i = 0; i < replIndex && neededReplicationsIterator.hasNext(); i++) {
neededReplicationsIterator.next();
}
Because our condition includes .hasNext() we will stop before trying to go outside our iterator. So we are safe with this change.
The reason we cannot decrementReplicationIndex(), or why it would be difficult to, is because we cannot guarantee that we will have actually removed the block from any SPECIFIC priority from within neededReplications. We do not know what the priority level of this block is within neededReplications when we try to do BlockManager.removeBlock(Block). If you look at remove(Block, int) in UnderReplicatedBlocks, it will try to remove it from the first queue, and if that fails, then tries to remove from all the queues. Therefore we will not be able to determine WHICH replicationIndex to decrement appropriately. However, as noted above, we are safe-guarded by the iterator nonetheless.
Now with that being said, I can understand from a consistency point as to why we would want to decrement the replicationIndex. However, in order to do it, we will need to add a method in UnderReplicatedBlocks such that we attempt to remove from a priority, and return true ONLY if we actually removed it from that priority; then we can safely decrement the replicationIndex of that priority.

Because our condition includes .hasNext() we will stop before trying to go outside our iterator. So we are safe with this change.

True! We won't throw an exception, but we would have skipped over a block. We won't get to this block again until the next iteration and a block with lower priority may be scheduled before a higher one. Its rather simple to fix this, so I would much rather that we did. (Although I noticed that there are already several instances in which we don't). Should we handle this in this JIRA / another?

Could we refactor UnderReplicatedBlocks so that decrementReplicationIndex is called from within remove()? I would be a +1 for that change.

Ravi Prakash
added a comment - 10/Jun/13 15:36 Hi Plamen! Thanks for your review.
Because our condition includes .hasNext() we will stop before trying to go outside our iterator. So we are safe with this change.
True! We won't throw an exception, but we would have skipped over a block. We won't get to this block again until the next iteration and a block with lower priority may be scheduled before a higher one. Its rather simple to fix this, so I would much rather that we did. (Although I noticed that there are already several instances in which we don't). Should we handle this in this JIRA / another?
Could we refactor UnderReplicatedBlocks so that decrementReplicationIndex is called from within remove()? I would be a +1 for that change.

Strictly speaking we should decrementReplicationIndex() if we remove the block at a position before the replicationIndex, and should not decrement if the block position is after the replicationIndex. In general we don't know the position of the block in the queue, so we cannot know whether the index should be decremented.
It should be targeted in a different jira if desired. The worst thing that happens if we don't change replicationIndex is that some blocks will be replicated later, which happens with some blocks one way or another. Eventually everything will be processed and that is what's important.
I'd rather commit the original patches.

Konstantin Shvachko
added a comment - 11/Jun/13 01:30 Strictly speaking we should decrementReplicationIndex() if we remove the block at a position before the replicationIndex, and should not decrement if the block position is after the replicationIndex. In general we don't know the position of the block in the queue, so we cannot know whether the index should be decremented.
It should be targeted in a different jira if desired. The worst thing that happens if we don't change replicationIndex is that some blocks will be replicated later, which happens with some blocks one way or another. Eventually everything will be processed and that is what's important.
I'd rather commit the original patches.