Repository: pinpoint-apm/pinpoint
Issue: Corrupted(Invalid Sequence) - Unable to trace DataStax Cassandra Batch Sql statement - CassandraStatementExecuteQueryInterceptor
Labels: bug
module:plugin

During in one of my transaction I am observing the Corrupted(Invalid Sequence) for couple of my transactions, when looking at the agent logs we had observed the below exception 


2017-06-27 15:07:50 [WARN ](draStatementExecuteQueryInterceptor) com.datastax.driver.core.BatchStatement cannot be cast to java.lang.String
java.lang.ClassCastException: com.datastax.driver.core.BatchStatement cannot be cast to java.lang.String
    at com.navercorp.pinpoint.plugin.cassandra.interceptor.CassandraStatementExecuteQueryInterceptor.before(CassandraStatementExecuteQueryInterceptor.java:102)
    at com.navercorp.pinpoint.bootstrap.interceptor.scope.ExceptionHandleScopedInterceptor.before(ExceptionHandleScopedInterceptor.java:57)
    at com.datastax.driver.core.AbstractSession.execute(AbstractSession.java)
    at com.target.oms.allocation.dao.NodeZonesDao.persist(NodeZonesDao.java:64)
    at com.target.oms.allocation.dao.NodeZonesDao.save(NodeZonesDao.java:53)
    at com.target.oms.allocation.dao.NodeZonesDao.save(NodeZonesDao.java:49)
    at com.target.oms.allocation.dao.NodeZonesDao$$FastClassBySpringCGLIB$$90186cce.invoke(<generated>


From the exception its pretty clear that Pinpoint agent does not trace the Cassandra Datastax api call for Batch Sql statements 

Specific code in CassandraStatementExecuteQueryInterceptor  here where we are observing the issue 

String sql;
            if (args[0] instanceof BoundStatement) {
                sql = ((BoundStatement) args[0]).preparedStatement().getQueryString();
            } else if (args[0] instanceof RegularStatement) {
                sql = ((RegularStatement) args[0]).getQueryString();
            } else {
                // we have string
                sql = (String) args[0];
            }

Kindly let me know If I have misinterpreted anything here 

Issue comments:
Looks like a bug in the Cassandra datastax plugin.
Thanks for reporting!

Pull request review comments:

Patch that resolved the issue:
diff --git a/plugins/cassandra/src/main/java/com/navercorp/pinpoint/plugin/cassandra/interceptor/CassandraStatementExecuteQueryInterceptor.java b/plugins/cassandra/src/main/java/com/navercorp/pinpoint/plugin/cassandra/interceptor/CassandraStatementExecuteQueryInterceptor.java
index dfeef8e456a9..551f0ca76bb5 100644
--- a/plugins/cassandra/src/main/java/com/navercorp/pinpoint/plugin/cassandra/interceptor/CassandraStatementExecuteQueryInterceptor.java
+++ b/plugins/cassandra/src/main/java/com/navercorp/pinpoint/plugin/cassandra/interceptor/CassandraStatementExecuteQueryInterceptor.java
@@ -19,8 +19,10 @@
 import java.util.HashMap;
 import java.util.Map;
 
+import com.datastax.driver.core.BatchStatement;
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.RegularStatement;
+import com.datastax.driver.core.StatementWrapper;
 import com.navercorp.pinpoint.bootstrap.context.DatabaseInfo;
 import com.navercorp.pinpoint.bootstrap.context.MethodDescriptor;
 import com.navercorp.pinpoint.bootstrap.context.ParsingResult;
@@ -85,33 +87,27 @@ public void before(Object target, Object[] args) {
             recorder.recordEndPoint(databaseInfo.getMultipleHost());
             recorder.recordDestinationId(databaseInfo.getDatabaseId());
 
-            String sql;
-            if (args[0] instanceof BoundStatement) {
-                sql = ((BoundStatement) args[0]).preparedStatement().getQueryString();
-            } else if (args[0] instanceof RegularStatement) {
-                sql = ((RegularStatement) args[0]).getQueryString();
-            } else {
-                // we have string
-                sql = (String) args[0];
-            }
+            String sql = retrieveSql(args[0]);
 
-            ParsingResult parsingResult = traceContext.parseSql(sql);
-            if (parsingResult != null) {
-                ((ParsingResultAccessor) target)._$PINPOINT$_setParsingResult(parsingResult);
-            } else {
-                if (logger.isErrorEnabled()) {
-                    logger.error("sqlParsing fail. parsingResult is null sql:{}", sql);
+            if (sql != null) {
+                ParsingResult parsingResult = traceContext.parseSql(sql);
+                if (parsingResult != null) {
+                    ((ParsingResultAccessor) target)._$PINPOINT$_setParsingResult(parsingResult);
+                } else {
+                    if (logger.isErrorEnabled()) {
+                        logger.error("sqlParsing fail. parsingResult is null sql:{}", sql);
+                    }
                 }
-            }
 
-            Map<Integer, String> bindValue = ((BindValueAccessor) target)._$PINPOINT$_getBindValue();
-            // TODO Add bind variable interceptors to BoundStatement's setter methods and bind method and pass it down
-            // Extracting bind variables from already-serialized is too risky
-            if (bindValue != null && !bindValue.isEmpty()) {
-                String bindString = toBindVariable(bindValue);
-                recorder.recordSqlParsingResult(parsingResult, bindString);
-            } else {
-                recorder.recordSqlParsingResult(parsingResult);
+                Map<Integer, String> bindValue = ((BindValueAccessor) target)._$PINPOINT$_getBindValue();
+                // TODO Add bind variable interceptors to BoundStatement's setter methods and bind method and pass it down
+                // Extracting bind variables from already-serialized is too risky
+                if (bindValue != null && !bindValue.isEmpty()) {
+                    String bindString = toBindVariable(bindValue);
+                    recorder.recordSqlParsingResult(parsingResult, bindString);
+                } else {
+                    recorder.recordSqlParsingResult(parsingResult);
+                }
             }
 
             recorder.recordApi(descriptor);
@@ -124,6 +120,26 @@ public void before(Object target, Object[] args) {
         }
     }
 
+    private String retrieveSql(Object args0) {
+        String sql;
+        if (args0 instanceof BoundStatement) {
+            sql = ((BoundStatement) args0).preparedStatement().getQueryString();
+        } else if (args0 instanceof RegularStatement) {
+            sql = ((RegularStatement) args0).getQueryString();
+        } else if (args0 instanceof StatementWrapper) {
+            // method to get wrapped statement is package-private, skip.
+            sql = null;
+        } else if (args0 instanceof BatchStatement) {
+            // we could unroll all the batched statements and append ; between them if need be but it could be too long.
+            sql = null;
+        } else if (args0 instanceof String) {
+            sql = (String) args0;
+        } else {
+            sql = null;
+        }
+        return sql;
+    }
+
     private void clean(Object target) {
         if (target instanceof BindValueAccessor) {
             ((BindValueAccessor) target)._$PINPOINT$_setBindValue(new HashMap<Integer, String>());


Test patch:
diff --git a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/cassandra/CassandraDatastaxITBase.java b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/cassandra/CassandraDatastaxITBase.java
index bc3cdb871887..c6f7550adf8a 100644
--- a/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/cassandra/CassandraDatastaxITBase.java
+++ b/agent/src/test/java/com/navercorp/pinpoint/plugin/jdk7/cassandra/CassandraDatastaxITBase.java
@@ -16,6 +16,7 @@
 
 package com.navercorp.pinpoint.plugin.jdk7.cassandra;
 
+import com.datastax.driver.core.BatchStatement;
 import com.datastax.driver.core.BoundStatement;
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.PreparedStatement;
@@ -23,6 +24,7 @@
 import com.datastax.driver.core.Row;
 import com.datastax.driver.core.Session;
 import com.datastax.driver.core.Statement;
+import com.datastax.driver.core.StatementWrapper;
 import com.navercorp.pinpoint.bootstrap.plugin.test.PluginTestVerifier;
 import com.navercorp.pinpoint.bootstrap.plugin.test.PluginTestVerifierHolder;
 import com.navercorp.pinpoint.common.util.DefaultSqlParser;
@@ -94,7 +96,7 @@ public static void initializeCluster(String cassandraVersion) throws Exception {
     }
 
     @Test
-    public void test() throws Exception {
+    public void testBoundStatement() throws Exception {
         final int testId = 99;
         final String testValue = "testValue";
 
@@ -157,6 +159,70 @@ public void test() throws Exception {
         }
     }
 
+    @Test
+    public void testBatchStatement_and_StatementWrapper() throws Exception {
+        final int testId1 = 998;
+        final String testValue1 = "testValue998";
+        final int testId2 = 999;
+        final String testValue2 = "testValue999";
+
+        PluginTestVerifier verifier = PluginTestVerifierHolder.getInstance();
+
+        Session myKeyspaceSession = null;
+
+        try {
+            myKeyspaceSession = cluster.connect(TEST_KEYSPACE);
+
+            // ===============================================
+            // Insert Data 2 x (PreparedStatement, BoundStatement)
+            PreparedStatement preparedStatement = myKeyspaceSession.prepare(CQL_INSERT);
+            BoundStatement boundStatement1 = new BoundStatement(preparedStatement);
+            boundStatement1.bind(testId1, testValue1);
+            BoundStatement boundStatement2 = new BoundStatement(preparedStatement);
+            boundStatement2.bind(testId2, testValue2);
+
+            BatchStatement batchStatement = new BatchStatement();
+            batchStatement.add(boundStatement1);
+            batchStatement.add(boundStatement2);
+
+            myKeyspaceSession.execute(batchStatement);
+
+            verifier.printCache();
+            // Cluster#connect(String)
+            Class<?> clusterClass = Class.forName("com.datastax.driver.core.Cluster");
+            Method connect = clusterClass.getDeclaredMethod("connect", String.class);
+            verifier.verifyTrace(event(CASSANDRA, connect, null, CASSANDRA_ADDRESS, TEST_KEYSPACE));
+            // SessionManager#prepare(String) OR AbstractSession#prepare(String)
+            Class<?> sessionClass;
+            try {
+                sessionClass = Class.forName("com.datastax.driver.core.AbstractSession");
+            } catch (ClassNotFoundException e) {
+                sessionClass = Class.forName("com.datastax.driver.core.SessionManager");
+            }
+            Method prepare = sessionClass.getDeclaredMethod("prepare", String.class);
+            verifier.verifyTrace(event(CASSANDRA, prepare, null, CASSANDRA_ADDRESS, TEST_KEYSPACE, sql(CQL_INSERT, null)));
+            // SessionManager#execute(Statement) OR AbstractSession#execute(Statement)
+            Method execute = sessionClass.getDeclaredMethod("execute", Statement.class);
+            verifier.verifyTrace(event(CASSANDRA_EXECUTE_QUERY, execute, null, CASSANDRA_ADDRESS, TEST_KEYSPACE));
+
+            // ====================
+            final String cqlDelete = String.format("DELETE FROM %s.%s WHERE %s IN (? , ?)", TEST_KEYSPACE, TEST_TABLE, TEST_COL_ID);
+            PreparedStatement deletePreparedStatement = myKeyspaceSession.prepare(cqlDelete);
+            BoundStatement deleteBoundStatement = new BoundStatement(deletePreparedStatement);
+            deleteBoundStatement.bind(testId1, testId2);
+            Statement wrappedDeleteStatement = new StatementWrapper(deleteBoundStatement) {};
+            myKeyspaceSession.execute(wrappedDeleteStatement);
+
+            verifier.printCache();
+            // SessionManager#prepare(String) OR AbstractSession#prepare(String)
+            verifier.verifyTrace(event(CASSANDRA, prepare, null, CASSANDRA_ADDRESS, TEST_KEYSPACE, sql(cqlDelete, null)));
+            // SessionManager#execute(String, Object[]) OR AbstractSession#execute(String, Object[])
+            verifier.verifyTrace(event(CASSANDRA_EXECUTE_QUERY, execute, null, CASSANDRA_ADDRESS, TEST_KEYSPACE));
+        } finally {
+            closeSession(myKeyspaceSession);
+        }
+    }
+
     private void verifySelect(ResultSet rs, int expectedTestId, String expectedTestValue) {
         int resultCount = 0;
         Iterator<Row> iter = rs.iterator();