Re: Exception on get method clusterj

[Arco van der Velden]

Hi Graig,

Setting exclusive locking did some magic, records are being updated well now but the exceptions still occur.
We're still getting:
"com.mysql.clusterj.ClusterJDatastoreException: Error in NdbJTie: returnCode -1, code 630, mysqlCode 121, status 2, classification 3, message Tuple already existed when attempting to insert" which is explainable because multiple threads try to insert the same record, this is resolved by retrying the transaction and no problem.
We also get:
"com.mysql.clusterj.ClusterJDatastoreException: For field total column total valueDelegate object BigDecimal, error executing objectGetValue. Caused by java.lang.IllegalStateException:Current state = FLUSHED, new state = CODING_END" which I don't understand, again due to the retry of the transaction at the end everything is ok but this doesn't feel ok.
The second exception can easily be reproduced by creating a simple table with
"CREATE TABLE `positions` (
`id` int(11) NOT NULL,
`total` decimal(30,7) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=ndbcluster DEFAULT CHARSET=utf8;"

and running the code below
"public class TestExceptions
{
private static final int numberOfThreads = 16;
public static ExecutorService executor = Executors.newCachedThreadPool();

public static void main(String[] args)
{
try {
//Setup factory
Properties props = new Properties();
props.put("com.mysql.clusterj.connectstring", "localhost:1186");
props.put("com.mysql.clusterj.database", "tps");
props.put("com.mysql.clusterj.connect.retries", "4");
props.put("com.mysql.clusterj.connect.delay", "5");
props.put("com.mysql.clusterj.connect.verbose", "1");
props.put("com.mysql.clusterj.connect.timeout.before", "30");
props.put("com.mysql.clusterj.connect.timeout.after", "20");
props.put("com.mysql.clusterj.max.transactions", "1024");
SessionFactory factory = ClusterJHelper.getSessionFactory(props);
//Clear table
Session session = factory.getSession();
session.currentTransaction().begin();
session.deletePersistentAll(PositionsInterface.class);
session.currentTransaction().commit();
session.close();
//Start threads
//ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 1; i <= numberOfThreads; i++) {
executor.execute(new WorkerThread(factory.getSession(), i));
}
executor.shutdown();
executor.awaitTermination(1, TimeUnit.DAYS);
}
catch (Exception e) {
e.printStackTrace();
}
}
@PersistenceCapable(table = "positions")
private interface PositionsInterface
{
//
//
@PrimaryKey
@Column(name = "id")
int getId();
void setId(int id);
@Column(name = "total")
BigDecimal getTotal();
void setTotal(BigDecimal total);
}
private static class WorkerThread implements Runnable
{
private final int threadNumber;

private final Random rand = new Random();

private final Session session;
public WorkerThread(Session session, int threadNumber)
{
this.session = session;
this.threadNumber = threadNumber;
}
@Override
public void run()
{
Thread.currentThread().setName("Worker-" + threadNumber);
processCommand();
}
private void processCommand()
{
try {
long start_time = System.currentTimeMillis();
session.setLockMode(LockMode.EXCLUSIVE);
//Do 5000 random inserts/updates
for (int i = 0; i < 5000; i++) {
//Process random records
int attempt = 1;
int maxAttempts = 5;
while (attempt <= maxAttempts) {
try {
session.currentTransaction().begin();
//Process record
int id = rand.nextInt((100 - 1) + 1) + 1;
PositionsInterface positionsInterface = session.find(PositionsInterface.class, id);
if (positionsInterface == null) {
//Insert
positionsInterface = session.newInstance(PositionsInterface.class, id);
positionsInterface.setTotal(new BigDecimal(BigInteger.ONE));
session.makePersistent(positionsInterface);
}
else {
//Update
positionsInterface.setTotal(positionsInterface.getTotal().add(new BigDecimal(BigInteger.ONE)));
session.updatePersistent(positionsInterface);
}
session.currentTransaction().commit();
break;
}
catch (ClusterJDatastoreException ex) {
if (session.currentTransaction().isActive()) {
session.currentTransaction().rollback();
}
System.out.println("ClusterJ datastore exception...");
ex.printStackTrace();
attempt++;
}
catch (Exception ex) {
System.out.println("General exception...");
ex.printStackTrace();
break;
}
}
}
System.out.println("It took:" + (System.currentTimeMillis() - start_time) + " ms");
}
catch (Exception ex) {
ex.printStackTrace();
}
}
}
}"