cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Scott McCarty (JIRA)" <j...@apache.org>
Subject [jira] Created: (CASSANDRA-1183) Column insert via batch_mutate under some conditions sometimes fails to add the column
Date Fri, 11 Jun 2010 16:32:14 GMT
Column insert via batch_mutate under some conditions sometimes fails to add the column
--------------------------------------------------------------------------------------

                 Key: CASSANDRA-1183
                 URL: https://issues.apache.org/jira/browse/CASSANDRA-1183
             Project: Cassandra
          Issue Type: Bug
          Components: Core
    Affects Versions: 0.6.1, 0.6.2
         Environment: OS (via uname -a):  Linux 2.6.27.5-117.fc10.x86_64 #1 SMP Tue Nov 18
11:58:53 EST 2008 x86_64 x86_64 x86_64 GNU/Linux
Java build 1.6.0_20-b02
Cassandra 0.6.2 & 0.6.1
hector-0.6.0-14.jar

            Reporter: Scott McCarty


Under heavy load batch_mutate sometimes fails to add a column but doesn't report an error
to the client.  Details:

* client is running a tight loop that does:  write to randomly named column; do a get_slice
that should return that newly created column; delete the column if it was returned in get_slice
* the mutation map for batch_insert has exactly one mutation in it:  insert the column
* if the column insert is done with the insert() API it doesn't ever fail (for as long as
I've run it)
* the column really isn't inserted as doing a 'get' on the entire rowkey in the command line
interface doesn't show it, so it's not a timing issue
* read consistency level is ONE, write consistency level is ALL
* it fails on a single node cluster and on a 4 node cluster (replication factor = 3)

I am including a Java cmdline program (below) that demonstrates the problem; it uses the 'hector'
java client (I hope that's okay--I verified that no errors or warning were being masked by
hector).  I didn't see a place to upload a file in bug reports so this program is in line.

To show the failure using the program, just run it with the command line:

  bash>   java BatchMutateBug localhost:9160 true keyspace columnfamily rowkey

substituting in the appropriate Cassandra server connection info, a valid keyspace name, and
a valid column family name.  The second parameter ("true" in the example) tells the program
to use batch_mutate().  If you pass in "false" it will use insert() and the program won't
fail.

I've tried the above on two different setups and each one fails within 10-15 seconds generally.
 It seems to fail quickest if you use a non-existing rowkey.

Here's the program:


import java.util.*;
import org.apache.cassandra.thrift.*;
import me.prettyprint.cassandra.service.*;

public class BatchMutateBug
{
    private static String[]  cHosts;
    private static String    cKeyspace;

    public static void main(
        String[] args
        )
    {
        BatchMutateBug c = new BatchMutateBug();

        if (args.length != 5)
            System.out.println("Usage:  BatchMutateBug host:port useBatchFlag keyspace columnfamily
rowkey");
        else
            c.contend(args[0], args[1], args[2], args[3], args[4]);
    }

    private void contend(
        String        hostPort,
        final String  batchFlag,
        String        keyspace,
        final String  cfname,
        final String  rowkey
        )
    {
        final  boolean  useBatch = Boolean.valueOf(batchFlag);
        Thread          wr1;

        cHosts = new String[] {hostPort};
        cKeyspace = keyspace;

        if (useBatch)
            System.out.println("Using batchMutate:  you should see an ERROR eventually");
        else
            System.out.println("NOT using batchMutate:  you should NOT see an ERROR");

        wr1 = new Thread()
            {
                public void run()
                {
                    try
                    {
                        writeRead(useBatch, cfname, rowkey);
                    }
                    catch (Exception x)
                    {
                        x.printStackTrace();
                    }
                }
            };

        wr1.start();
    }

    /**
     *
     */
    private void writeRead(
        boolean useBatch,
        String  cfname,
        String  rowkey
        )
        throws Exception
    {
        Random       rand   = new Random();
        ColumnParent par    = new ColumnParent(cfname);
        ColumnPath   cp     = new ColumnPath(cfname);
        Keyspace     ks;
        String       colname;
        String       rpart;

        while (true)
        {
            boolean  pass = false;

            ks      = getWriteKeyspace();
            rpart   = Integer.toString(rand.nextInt());
            colname = "%reader:" + rpart;

            try
            {
                if (useBatch)
                {
                    Map<String, Map<String, List<Mutation>>> muts = new
HashMap<String, Map<String, List<Mutation>>>();
                    Map<String, List<Mutation>>              cfm  = new HashMap<String,
List<Mutation>>();
                    List<Mutation>                           mutl = new ArrayList<Mutation>();
                    Mutation                                 mut  = new Mutation();

                    mutl.add(mut);
                    cfm.put(cfname, mutl);
                    muts.put(rowkey, cfm);

                    mut.column_or_supercolumn = new ColumnOrSuperColumn();
                    mut.column_or_supercolumn.column = new Column();
                    mut.column_or_supercolumn.column.name = colname.getBytes();
                    mut.column_or_supercolumn.column.value = colname.getBytes();
                    mut.column_or_supercolumn.column.timestamp = System.currentTimeMillis()
* 1000L;

                    ks.batchMutate(muts);
                }
                else
                {
                    cp.column = colname.getBytes();
                    ks.insert(rowkey, cp, colname.getBytes());
                }
            }
            finally
            {
                releaseKeyspace(ks);
            }

            ks = getReadKeyspace();

            try
            {
                List<Column>   cols = readSlice(rowkey, cfname, "%", "&");

                if (cols.size() == 0)
                {
                    System.out.println("ERROR:  column '" + colname + "' was inserted but
getSlice returned 0 columns");
                }
                else
                {
                    boolean found = false;

                    for (Column c : cols)
                    {
                        String  fetched = new String(c.name);

                        if (colname.equals(fetched))
                        {
                            found = true;
                            pass = true;
                            break;
                        }
                    }

                    if (!found)
                        System.out.println("ERROR:  column '" + colname + "' was inserted
but getSlice did NOT return it");
                }
            }
            finally
            {
                releaseKeyspace(ks);
            }

            ks = getWriteKeyspace();

            if (pass)
            {
                try
                {
                    ks.remove(rowkey, cp);
                }
                finally
                {
                    releaseKeyspace(ks);
                }
            }
        }
    }

    private List<Column> readSlice(
        String    rowkey,
        String    cfname,
        String    start,
        String    finish
        )
        throws Exception
    {
        Keyspace      ks   = getReadKeyspace();
        List<Column>  cols = null;
        ColumnParent  par  = new ColumnParent(cfname);

        try
        {
            SlicePredicate sp;

            sp = new SlicePredicate();
            sp.slice_range = new SliceRange();
            sp.slice_range.count = 100;
            sp.slice_range.start = start.getBytes();
            sp.slice_range.finish = finish.getBytes();

            cols = ks.getSlice(rowkey, par, sp);
        }
        catch (NotFoundException nfe)
        {
            System.out.println("ERROR:  got NotFoundException");
        }
        finally
        {
            releaseKeyspace(ks);
        }

        return cols;
    }

    private Keyspace getReadKeyspace()
    {
        try
        {
            CassandraClientPool ccp = CassandraClientPoolFactory.INSTANCE.get();
            CassandraClient     c   = ccp.borrowClient(cHosts);
            me.prettyprint.cassandra.service.Keyspace ks;

            ks = c.getKeyspace(cKeyspace, ConsistencyLevel.ONE);

            return ks;
        }
        catch (Exception x)
        {
            x.printStackTrace();
        }

        return null;
    }

    private Keyspace getWriteKeyspace()
    {
        try
        {
            CassandraClientPool ccp = CassandraClientPoolFactory.INSTANCE.get();
            CassandraClient     c   = ccp.borrowClient(cHosts);
            me.prettyprint.cassandra.service.Keyspace ks;

            ks = c.getKeyspace(cKeyspace, ConsistencyLevel.ALL);

            return ks;
        }
        catch (Exception x)
        {
            x.printStackTrace();
        }

        return null;
    }

    protected static void releaseKeyspace(
        Keyspace  ks
        )
    {
        if (ks != null)
        {
            try
            {
                CassandraClientPool ccp = CassandraClientPoolFactory.INSTANCE.get();

                ccp.releaseClient(ks.getClient());
            }
            catch (Exception x)
            {
                x.printStackTrace();
            }
        }
    }

}


-- 
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.


Mime
View raw message