src/java/org/apache/cassandra/db/context/CounterContext.java
/**
     * Compute a new context such that if applied to context yields the same
     * total but with old local node ids nulified and there content merged to
     * the current localNodeId.
     */
Compute a new context such that if applied to context yields the same
total but with old local node ids nulified and there content merged to
the current localNodeId.
Compute a new context such that if applied to context yields the same total but with old local node ids nulified and there content merged to the current localNodeId .
public ByteBuffer computeOldShardMerger(ByteBuffer context, List<NodeId.NodeIdRecord> oldIds, long mergeBefore)
    {
        long now = System.currentTimeMillis();
        int hlength = headerLength(context);
        NodeId localId = NodeId.getLocalId();
        Iterator<NodeId.NodeIdRecord> recordIterator = oldIds.iterator();
        NodeId.NodeIdRecord currRecord = recordIterator.hasNext() ? recordIterator.next() : null;
        ContextState state = new ContextState(context, hlength);
        ContextState foundState = null;
        List<NodeId> toMerge = new ArrayList<NodeId>();
        long mergeTotal = 0;
        while (state.hasRemaining() && currRecord != null)
        {
            assert !currRecord.id.equals(localId);
            NodeId nodeId = state.getNodeId();
            int c = nodeId.compareTo(currRecord.id);
            if (c > 0)
            {
                currRecord = recordIterator.hasNext() ? recordIterator.next() : null;
                continue;
            }
            if (state.isDelta())
            {
                if (state.getClock() < 0)
                {
                    if (nodeId.equals(localId))
                        throw new RuntimeException("Current nodeId with a negative clock (likely due to #2968). You need to restart this node with -Dcassandra.renew_counter_id=true to fix.");
                    if (state.getCount() != 0)
                    {
                        logger.error(String.format("Invalid counter context (clock is %d and count is %d for NodeId %s), will fix", state.getCount(), state.getCount(), nodeId.toString()));
                        toMerge.add(nodeId);
                        mergeTotal += state.getCount();
                    }
                }
                else if (c == 0)
                {
                    if (currRecord.timestamp < mergeBefore)
                    {
                        toMerge.add(nodeId);
                        mergeTotal += state.getCount();
                    }
                }
            }
            if (c == 0)
                currRecord = recordIterator.hasNext() ? recordIterator.next() : null;
            state.moveToNext();
        }
        while (state.hasRemaining())
        {
            NodeId nodeId = state.getNodeId();
            if (state.isDelta() && state.getClock() < 0)
            {
                if (nodeId.equals(localId))
                    throw new RuntimeException("Current nodeId with a negative clock (likely due to #2968). You need to restart this node with -Dcassandra.renew_counter_id=true to fix.");
                if (state.getCount() != 0)
                {
                    logger.error(String.format("Invalid counter context (clock is %d and count is %d for NodeId %s), will fix", state.getClock(), state.getCount(), nodeId.toString()));
                    toMerge.add(nodeId);
                    mergeTotal += state.getCount();
                }
            }
            state.moveToNext();
        }
        if (toMerge.isEmpty())
            return null;
        ContextState merger = ContextState.allocate(toMerge.size() + 1, toMerge.size() + 1);
        state.reset();
        int i = 0;
        int removedTotal = 0;
        boolean localWritten = false;
        while (state.hasRemaining())
        {
            NodeId nodeId = state.getNodeId();
            if (nodeId.compareTo(localId) > 0)
            {
                merger.writeElement(localId, 1L, mergeTotal, LamportClock.getVersion(), ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
                localWritten = true;
            }
            else if (i < toMerge.size() && nodeId.compareTo(toMerge.get(i)) == 0)
            {
                long count = state.getCount();
                removedTotal += count;
                merger.writeElement(nodeId, -now - state.getClock(), -count, state.getTimestamp(), ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
                ++i;
            }
            state.moveToNext();
        }
        if (!localWritten)
            merger.writeElement(localId, 1L, mergeTotal, LamportClock.getVersion(), ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
        assert mergeTotal == removedTotal;
        return merger.context;
    }
public ByteBuffer computeOldShardMerger ( ByteBuffer context , List < NodeId . NodeIdRecord > oldIds , long mergeBefore ) { long now = System . currentTimeMillis ( ) ; int hlength = headerLength ( context ) ; NodeId localId = NodeId . getLocalId ( ) ; Iterator < NodeId . NodeIdRecord > recordIterator = oldIds . iterator ( ) ; NodeId . NodeIdRecord currRecord = recordIterator . hasNext ( ) ? recordIterator . next ( ) : null ; ContextState state = new ContextState ( context , hlength ) ; ContextState foundState = null ; List < NodeId > toMerge = new ArrayList < NodeId > ( ) ; long mergeTotal = 0 ; while ( state . hasRemaining ( ) && currRecord != null ) { assert ! currRecord . id . equals ( localId ) ; NodeId nodeId = state . getNodeId ( ) ; int c = nodeId . compareTo ( currRecord . id ) ; if ( c > 0 ) { currRecord = recordIterator . hasNext ( ) ? recordIterator . next ( ) : null ; continue ; } if ( state . isDelta ( ) ) { if ( state . getClock ( ) < 0 ) { if ( nodeId . equals ( localId ) ) throw new RuntimeException ( "Current nodeId with a negative clock (likely due to #2968). You need to restart this node with -Dcassandra.renew_counter_id=true to fix." ) ; if ( state . getCount ( ) != 0 ) { logger . error ( String . format ( "Invalid counter context (clock is %d and count is %d for NodeId %s), will fix" , state . getCount ( ) , state . getCount ( ) , nodeId . toString ( ) ) ) ; toMerge . add ( nodeId ) ; mergeTotal += state . getCount ( ) ; } } else if ( c == 0 ) { if ( currRecord . timestamp < mergeBefore ) { toMerge . add ( nodeId ) ; mergeTotal += state . getCount ( ) ; } } } if ( c == 0 ) currRecord = recordIterator . hasNext ( ) ? recordIterator . next ( ) : null ; state . moveToNext ( ) ; } while ( state . hasRemaining ( ) ) { NodeId nodeId = state . getNodeId ( ) ; if ( state . isDelta ( ) && state . getClock ( ) < 0 ) { if ( nodeId . equals ( localId ) ) throw new RuntimeException ( "Current nodeId with a negative clock (likely due to #2968). You need to restart this node with -Dcassandra.renew_counter_id=true to fix." ) ; if ( state . getCount ( ) != 0 ) { logger . error ( String . format ( "Invalid counter context (clock is %d and count is %d for NodeId %s), will fix" , state . getClock ( ) , state . getCount ( ) , nodeId . toString ( ) ) ) ; toMerge . add ( nodeId ) ; mergeTotal += state . getCount ( ) ; } } state . moveToNext ( ) ; } if ( toMerge . isEmpty ( ) ) return null ; ContextState merger = ContextState . allocate ( toMerge . size ( ) + 1 , toMerge . size ( ) + 1 ) ; state . reset ( ) ; int i = 0 ; int removedTotal = 0 ; boolean localWritten = false ; while ( state . hasRemaining ( ) ) { NodeId nodeId = state . getNodeId ( ) ; if ( nodeId . compareTo ( localId ) > 0 ) { merger . writeElement ( localId , 1L , mergeTotal , LamportClock . getVersion ( ) , ByteBufferUtil . EMPTY_BYTE_BUFFER , true ) ; localWritten = true ; } else if ( i < toMerge . size ( ) && nodeId . compareTo ( toMerge . get ( i ) ) == 0 ) { long count = state . getCount ( ) ; removedTotal += count ; merger . writeElement ( nodeId , - now - state . getClock ( ) , - count , state . getTimestamp ( ) , ByteBufferUtil . EMPTY_BYTE_BUFFER , true ) ; ++ i ; } state . moveToNext ( ) ; } if ( ! localWritten ) merger . writeElement ( localId , 1L , mergeTotal , LamportClock . getVersion ( ) , ByteBufferUtil . EMPTY_BYTE_BUFFER , true ) ; assert mergeTotal == removedTotal ; return merger . context ; }
Compute a new context such that if applied to context yields the same
 total but with old local node ids nulified and there content merged to
 the current localNodeId.
Compute a new context such that if applied to context yields the same total but with old local node ids nulified and there content merged to the current localNodeId .
// Already merged shard, waiting to be collected // we should not get there, but we have been creating problematic context prior to #2968 // This should not happen, but previous bugs have generated this (#2968 in particular) so fixing it. // Found an old id. However, merging an oldId that has just been renewed isn't safe, so // we check that it has been renewed before mergeBefore. // Continuing the iteration so that we can repair invalid shards // we should not get there, but we have been creating problematic context prior to #2968 // This should not happen, but previous bugs have generated this (#2968 in particular) so fixing it. // sanity check