Make BatchStatementBuilder.build log warning if statement with (#325)

non-default consistancy level is in the batch

When user does:
```
  SimpleStatement simpleStatement = SimpleStatement.builder("SELECT * FROM some_table WHERE a = ?")
    .setConsistencyLevel(DefaultConsistencyLevel.QUORUM).build();
  batchStatementBuilder.addStatement(simpleStatement);
  batchStatementBuilder.build();
```

Child statement looses it's consistancy level, which is confusing.
To warn user it is happening driver will issue log warning.

Author: dkropachev

core/src/main/java/com/datastax/oss/driver/internal/core/cql/DefaultBatchStatement.java

@@ -42,9 +42,12 @@
 import java.util.List;
 import java.util.Map;
 import net.jcip.annotations.Immutable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Immutable
 public class DefaultBatchStatement implements BatchStatement {
+  private static final Logger LOG = LoggerFactory.getLogger(DefaultBatchStatement.class);
 
   private final BatchType batchType;
   private final List<BatchableStatement<?>> statements;
@@ -86,6 +89,18 @@ public DefaultBatchStatement(
       Duration timeout,
       Node node,
       int nowInSeconds) {
+    for (BatchableStatement<?> statement : statements) {
+      if (statement != null
+          && (statement.getConsistencyLevel() != null
+              || statement.getSerialConsistencyLevel() != null)) {
+
+        LOG.warn(
+            "You have submitted statement with non-default [serial] consistency level to the DefaultBatchStatement. "
+                + "Be aware that [serial] consistency level of child statements is not preserved by the DefaultBatchStatement. "
+                + "Use DefaultBatchStatement.setConsistencyLevel()/DefaultBatchStatement.setSerialConsistencyLevel() instead.");
+        break;
+      }
+    }
     this.batchType = batchType;
     this.statements = ImmutableList.copyOf(statements);
     this.executionProfileName = executionProfileName;

core/src/test/java/com/datastax/oss/driver/internal/core/cql/DefaultBatchStatementTest.java

@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datastax.oss.driver.internal.core.cql;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
+import com.datastax.oss.driver.api.core.cql.BatchStatementBuilder;
+import com.datastax.oss.driver.api.core.cql.BatchType;
+import com.datastax.oss.driver.api.core.cql.SimpleStatement;
+import com.datastax.oss.driver.internal.core.util.LoggerTest;
+import org.junit.Test;
+
+public class DefaultBatchStatementTest {
+
+  @Test
+  public void should_issue_log_warn_if_statement_have_consistency_level_set() {
+    SimpleStatement simpleStatement =
+        SimpleStatement.builder("SELECT * FROM some_table WHERE a = ?")
+            .setConsistencyLevel(DefaultConsistencyLevel.QUORUM)
+            .build();
+
+    BatchStatementBuilder batchStatementBuilder = new BatchStatementBuilder(BatchType.LOGGED);
+    batchStatementBuilder.addStatement(simpleStatement);
+
+    LoggerTest.LoggerSetup logger =
+        LoggerTest.setupTestLogger(DefaultBatchStatement.class, Level.WARN);
+
+    batchStatementBuilder.build();
+
+    verify(logger.appender).doAppend(logger.loggingEventCaptor.capture());
+    assertThat(
+            logger.loggingEventCaptor.getAllValues().stream()
+                .map(ILoggingEvent::getFormattedMessage))
+        .contains(
+            "You have submitted statement with non-default [serial] consistency level to the DefaultBatchStatement. "
+                + "Be aware that [serial] consistency level of child statements is not preserved by the DefaultBatchStatement. "
+                + "Use DefaultBatchStatement.setConsistencyLevel()/DefaultBatchStatement.setSerialConsistencyLevel() instead.");
+  }
+
+  @Test
+  public void should_issue_log_warn_if_statement_have_serial_consistency_level_set() {
+    SimpleStatement simpleStatement =
+        SimpleStatement.builder("SELECT * FROM some_table WHERE a = ?")
+            .setSerialConsistencyLevel(DefaultConsistencyLevel.LOCAL_SERIAL)
+            .build();
+
+    BatchStatementBuilder batchStatementBuilder = new BatchStatementBuilder(BatchType.LOGGED);
+    batchStatementBuilder.addStatement(simpleStatement);
+
+    LoggerTest.LoggerSetup logger =
+        LoggerTest.setupTestLogger(DefaultBatchStatement.class, Level.WARN);
+
+    batchStatementBuilder.build();
+
+    verify(logger.appender).doAppend(logger.loggingEventCaptor.capture());
+    assertThat(
+            logger.loggingEventCaptor.getAllValues().stream()
+                .map(ILoggingEvent::getFormattedMessage))
+        .contains(
+            "You have submitted statement with non-default [serial] consistency level to the DefaultBatchStatement. "
+                + "Be aware that [serial] consistency level of child statements is not preserved by the DefaultBatchStatement. "
+                + "Use DefaultBatchStatement.setConsistencyLevel()/DefaultBatchStatement.setSerialConsistencyLevel() instead.");
+  }
+
+  @Test
+  public void should_not_issue_log_warn_if_statement_have_no_consistency_level_set() {
+    SimpleStatement simpleStatement =
+        SimpleStatement.builder("SELECT * FROM some_table WHERE a = ?").build();
+
+    BatchStatementBuilder batchStatementBuilder = new BatchStatementBuilder(BatchType.LOGGED);
+    batchStatementBuilder.addStatement(simpleStatement);
+
+    LoggerTest.LoggerSetup logger =
+        LoggerTest.setupTestLogger(DefaultBatchStatement.class, Level.WARN);
+
+    batchStatementBuilder.build();
+
+    verify(logger.appender, times(0)).doAppend(logger.loggingEventCaptor.capture());
+  }
+}

integration-tests/src/test/java/com/datastax/oss/driver/core/cql/BatchStatementIT.java

@@ -25,7 +25,11 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.verify;
 
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
@@ -45,6 +49,8 @@
 import com.datastax.oss.driver.api.testinfra.session.SessionRule;
 import com.datastax.oss.driver.api.testinfra.session.SessionUtils;
 import com.datastax.oss.driver.categories.ParallelizableTests;
+import com.datastax.oss.driver.internal.core.cql.DefaultBatchStatement;
+import com.datastax.oss.driver.internal.core.util.LoggerTest;
 import java.util.Iterator;
 import java.util.List;
 import org.junit.Before;
@@ -88,6 +94,32 @@ public void createTable() {
     }
   }
 
+  @Test
+  public void should_issue_log_warn_if_batched_statement_have_consistency_level_set() {
+    SimpleStatement simpleStatement =
+        SimpleStatement.builder("INSERT INTO test (k0, k1, v) values ('123123', ?, ?)").build();
+
+    try (CqlSession session = SessionUtils.newSession(ccmRule, sessionRule.keyspace())) {
+      PreparedStatement prep = session.prepare(simpleStatement);
+      BatchStatementBuilder batch = BatchStatement.builder(DefaultBatchType.UNLOGGED);
+      batch.addStatement(prep.bind(1, 2).setConsistencyLevel(ConsistencyLevel.QUORUM));
+
+      LoggerTest.LoggerSetup logger =
+          LoggerTest.setupTestLogger(DefaultBatchStatement.class, Level.WARN);
+
+      batch.build();
+
+      verify(logger.appender).doAppend(logger.loggingEventCaptor.capture());
+      assertThat(
+              logger.loggingEventCaptor.getAllValues().stream()
+                  .map(ILoggingEvent::getFormattedMessage))
+          .contains(
+              "You have submitted statement with non-default [serial] consistency level to the DefaultBatchStatement. "
+                  + "Be aware that [serial] consistency level of child statements is not preserved by the DefaultBatchStatement. "
+                  + "Use DefaultBatchStatement.setConsistencyLevel()/DefaultBatchStatement.setSerialConsistencyLevel() instead.");
+    }
+  }
+
   @Test
   public void should_execute_batch_of_simple_statements_with_variables() {
     // Build a batch of batchCount simple statements, each with their own positional variables.