Merge branch '4.11.x' into 4.x

Author: adutra

Jenkinsfile

@@ -220,8 +220,7 @@ pipeline {
                 'dse-5.1',   // Legacy DataStax Enterprise
                 'dse-6.0',   // Previous DataStax Enterprise
                 'dse-6.7',   // Previous DataStax Enterprise
-                'dse-6.8.0', // Current DataStax Enterprise
-                'dse-6.8',   // Development DataStax Enterprise
+                'dse-6.8',   // Current DataStax Enterprise
                 'ALL'],
       description: '''Apache Cassandra&reg; and DataStax Enterprise server version to use for adhoc <b>BUILD-AND-EXECUTE-TESTS</b> builds
                       <table style="width:100%">
@@ -271,13 +270,9 @@ pipeline {
                           <td><strong>dse-6.7</strong></td>
                           <td>DataStax Enterprise v6.7.x</td>
                         </tr>
-                        <tr>
-                          <td><strong>dse-6.8.0</strong></td>
-                          <td>DataStax Enterprise v6.8.0</td>
-                        </tr>
                         <tr>
                           <td><strong>dse-6.8</strong></td>
-                          <td>DataStax Enterprise v6.8.x (<b>CURRENTLY UNDER DEVELOPMENT</b>)</td>
+                          <td>DataStax Enterprise v6.8.x</td>
                         </tr>
                       </table>''')
     choice(
@@ -361,13 +356,13 @@ pipeline {
     // schedules only run against release branches (i.e. 3.x, 4.x, 4.5.x, etc.)
     parameterizedCron(branchPatternCron.matcher(env.BRANCH_NAME).matches() ? """
       # Every weeknight (Monday - Friday) around 2:00 AM
-      ### JDK8 tests against 2.1, 3.0, DSE 4.8, DSE 5.0, DSE 5.1, DSE-6.0 and DSE 6.7
-      H 2 * * 1-5 %CI_SCHEDULE=WEEKNIGHTS;CI_SCHEDULE_SERVER_VERSIONS=2.1 3.0 dse-4.8 dse-5.0 dse-5.1 dse-6.0 dse-6.7;CI_SCHEDULE_JABBA_VERSION=1.8
-      ### JDK11 tests against 3.11, 4.0, DSE 6.7 and DSE 6.8.0
-      H 2 * * 1-5 %CI_SCHEDULE=WEEKNIGHTS;CI_SCHEDULE_SERVER_VERSIONS=3.11 4.0 dse-6.7 dse-6.8.0;[email protected]
+      ### JDK8 tests against 2.1, 3.0, DSE 4.8, DSE 5.0, DSE 5.1, DSE-6.0, DSE 6.7 and DSE 6.8
+      H 2 * * 1-5 %CI_SCHEDULE=WEEKNIGHTS;CI_SCHEDULE_SERVER_VERSIONS=2.1 3.0 dse-4.8 dse-5.0 dse-5.1 dse-6.0 dse-6.7 dse-6.8;CI_SCHEDULE_JABBA_VERSION=1.8
+      ### JDK11 tests against 3.11, 4.0 and DSE 6.8
+      H 2 * * 1-5 %CI_SCHEDULE=WEEKNIGHTS;CI_SCHEDULE_SERVER_VERSIONS=3.11 4.0 dse-6.8;[email protected]
       # Every weekend (Sunday) around 12:00 PM noon
-      ### JDK14 tests against 3.11, 4.0, DSE 6.7, DSE 6.8.0 and DSE 6.8.X
-      H 12 * * 0 %CI_SCHEDULE=WEEKENDS;CI_SCHEDULE_SERVER_VERSIONS=3.11 4.0 dse-6.7 dse-6.8.0 dse-6.8;[email protected]
+      ### JDK14 tests against 3.11, 4.0 and DSE 6.8
+      H 12 * * 0 %CI_SCHEDULE=WEEKENDS;CI_SCHEDULE_SERVER_VERSIONS=3.11 4.0 dse-6.8;[email protected]
     """ : "")
   }
 
@@ -403,7 +398,7 @@ pipeline {
             name 'SERVER_VERSION'
             values '3.11',     // Latest stable Apache CassandraⓇ
                    '4.0',      // Development Apache CassandraⓇ
-                   'dse-6.8.0' // Current DataStax Enterprise
+                   'dse-6.8' // Current DataStax Enterprise
           }
         }
 
@@ -521,8 +516,7 @@ pipeline {
                    'dse-5.1',   // Legacy DataStax Enterprise
                    'dse-6.0',   // Previous DataStax Enterprise
                    'dse-6.7',   // Previous DataStax Enterprise
-                   'dse-6.8.0', // Current DataStax Enterprise
-                   'dse-6.8'    // Development DataStax Enterprise
+                   'dse-6.8'    // Current DataStax Enterprise
           }
         }
         when {

changelog/README.md

@@ -5,6 +5,7 @@
 ### 4.11.2 (in progress)
 
 - [bug] JAVA-2945: Reinstate InternalDriverContext.getNodeFilter method
+- [bug] JAVA-2947: Release buffer after decoding multi-slice frame
 - [bug] JAVA-2946: Make MapperResultProducerService instances be located with user-provided class loader
 - [bug] JAVA-2942: GraphStatement.setConsistencyLevel() is not effective
 - [bug] JAVA-2941: Cannot add a single static column with the alter table API

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java

@@ -314,6 +314,7 @@ private class SingleThreaded {
     private SingleThreaded(InternalDriverContext context, DriverExecutionProfile config) {
       this.schemaRefreshDebouncer =
           new Debouncer<>(
+              logPrefix + "|metadata debouncer",
               adminExecutor,
               this::coalesceSchemaRequests,
               this::startSchemaRequest,

core/src/main/java/com/datastax/oss/driver/internal/core/metadata/NodeStateManager.java

@@ -103,6 +103,7 @@ private SingleThreaded(InternalDriverContext context) {
       DriverExecutionProfile config = context.getConfig().getDefaultProfile();
       this.topologyEventDebouncer =
           new Debouncer<>(
+              logPrefix + "|topology debouncer",
               adminExecutor,
               this::coalesceTopologyEvents,
               this::flushTopologyEvents,

core/src/main/java/com/datastax/oss/driver/internal/core/protocol/SegmentToFrameDecoder.java

@@ -69,14 +69,19 @@ protected void decode(
   private void decodeSelfContained(Segment<ByteBuf> segment, List<Object> out) {
     ByteBuf payload = segment.payload;
     int frameCount = 0;
-    do {
-      Frame frame = frameCodec.decode(payload);
-      LOG.trace(
-          "[{}] Decoded response frame {} from self-contained segment", logPrefix, frame.streamId);
-      out.add(frame);
-      frameCount += 1;
-    } while (payload.isReadable());
-    payload.release();
+    try {
+      do {
+        Frame frame = frameCodec.decode(payload);
+        LOG.trace(
+            "[{}] Decoded response frame {} from self-contained segment",
+            logPrefix,
+            frame.streamId);
+        out.add(frame);
+        frameCount += 1;
+      } while (payload.isReadable());
+    } finally {
+      payload.release();
+    }
     LOG.trace("[{}] Done processing self-contained segment ({} frames)", logPrefix, frameCount);
   }
 
@@ -89,28 +94,34 @@ private void decodeSlice(Segment<ByteBuf> segment, ByteBufAllocator allocator, L
     }
     accumulatedSlices.add(slice);
     accumulatedLength += slice.readableBytes();
+    int accumulatedSlicesSize = accumulatedSlices.size();
     LOG.trace(
         "[{}] Decoded slice {}, {}/{} bytes",
         logPrefix,
-        accumulatedSlices.size(),
+        accumulatedSlicesSize,
         accumulatedLength,
         targetLength);
     assert accumulatedLength <= targetLength;
     if (accumulatedLength == targetLength) {
       // We've received enough data to reassemble the whole message
-      CompositeByteBuf encodedFrame = allocator.compositeBuffer(accumulatedSlices.size());
+      CompositeByteBuf encodedFrame = allocator.compositeBuffer(accumulatedSlicesSize);
       encodedFrame.addComponents(true, accumulatedSlices);
-      Frame frame = frameCodec.decode(encodedFrame);
+      Frame frame;
+      try {
+        frame = frameCodec.decode(encodedFrame);
+      } finally {
+        encodedFrame.release();
+        // Reset our state
+        targetLength = UNKNOWN_LENGTH;
+        accumulatedSlices.clear();
+        accumulatedLength = 0;
+      }
       LOG.trace(
           "[{}] Decoded response frame {} from {} slices",
           logPrefix,
           frame.streamId,
-          accumulatedSlices.size());
+          accumulatedSlicesSize);
       out.add(frame);
-      // Reset our state
-      targetLength = UNKNOWN_LENGTH;
-      accumulatedSlices.clear();
-      accumulatedLength = 0;
     }
   }
 }

core/src/main/java/com/datastax/oss/driver/internal/core/util/concurrent/Debouncer.java

@@ -44,6 +44,7 @@
 public class Debouncer<IncomingT, CoalescedT> {
   private static final Logger LOG = LoggerFactory.getLogger(Debouncer.class);
 
+  private final String logPrefix;
   private final EventExecutor adminExecutor;
   private final Consumer<CoalescedT> onFlush;
   private final Duration window;
@@ -69,6 +70,27 @@ public Debouncer(
       Consumer<CoalescedT> onFlush,
       Duration window,
       long maxEvents) {
+    this("debouncer", adminExecutor, coalescer, onFlush, window, maxEvents);
+  }
+
+  /**
+   * Creates a new instance.
+   *
+   * @param logPrefix the log prefix to use in log messages.
+   * @param adminExecutor the executor that will be used to schedule all tasks.
+   * @param coalescer how to transform a batch of events into a result.
+   * @param onFlush what to do with a result.
+   * @param window the time window.
+   * @param maxEvents the maximum number of accumulated events before a flush is forced.
+   */
+  public Debouncer(
+      String logPrefix,
+      EventExecutor adminExecutor,
+      Function<List<IncomingT>, CoalescedT> coalescer,
+      Consumer<CoalescedT> onFlush,
+      Duration window,
+      long maxEvents) {
+    this.logPrefix = logPrefix;
     this.coalescer = coalescer;
     Preconditions.checkArgument(maxEvents >= 1, "maxEvents should be at least 1");
     this.adminExecutor = adminExecutor;
@@ -85,7 +107,8 @@ public void receive(IncomingT element) {
     }
     if (window.isZero() || maxEvents == 1) {
       LOG.debug(
-          "Received {}, flushing immediately (window = {}, maxEvents = {})",
+          "[{}] Received {}, flushing immediately (window = {}, maxEvents = {})",
+          logPrefix,
           element,
           window,
           maxEvents);
@@ -94,20 +117,21 @@ public void receive(IncomingT element) {
       currentBatch.add(element);
       if (currentBatch.size() == maxEvents) {
         LOG.debug(
-            "Received {}, flushing immediately (because {} accumulated events)",
+            "[{}] Received {}, flushing immediately (because {} accumulated events)",
+            logPrefix,
             element,
             maxEvents);
         flushNow();
       } else {
-        LOG.debug("Received {}, scheduling next flush in {}", element, window);
+        LOG.debug("[{}] Received {}, scheduling next flush in {}", logPrefix, element, window);
         scheduleFlush();
       }
     }
   }
 
   public void flushNow() {
     assert adminExecutor.inEventLoop();
-    LOG.debug("Flushing now");
+    LOG.debug("[{}] Flushing now", logPrefix);
     cancelNextFlush();
     if (!currentBatch.isEmpty()) {
       onFlush.accept(coalescer.apply(currentBatch));
@@ -127,7 +151,7 @@ private void cancelNextFlush() {
     if (nextFlush != null && !nextFlush.isDone()) {
       boolean cancelled = nextFlush.cancel(true);
       if (cancelled) {
-        LOG.debug("Cancelled existing scheduled flush");
+        LOG.debug("[{}] Cancelled existing scheduled flush", logPrefix);
       }
     }
   }

core/src/test/java/com/datastax/oss/driver/internal/core/protocol/SegmentToFrameDecoderTest.java

@@ -73,7 +73,8 @@ public void should_decode_sequence_of_slices() {
         encodeFrame(new AuthResponse(Bytes.fromHexString("0x" + Strings.repeat("aa", 1011))));
     int sliceLength = 100;
     do {
-      ByteBuf payload = encodedFrame.readSlice(Math.min(sliceLength, encodedFrame.readableBytes()));
+      ByteBuf payload =
+          encodedFrame.readRetainedSlice(Math.min(sliceLength, encodedFrame.readableBytes()));
       channel.writeInbound(new Segment<>(payload, false));
     } while (encodedFrame.isReadable());
 

integration-tests/pom.xml

@@ -254,6 +254,7 @@
               <reuseForks>false</reuseForks>
               <summaryFile>${project.build.directory}/failsafe-reports/failsafe-summary-isolated.xml</summaryFile>
               <skipITs>${skipIsolatedITs}</skipITs>
+              <argLine>${blockhound.argline}</argLine>
             </configuration>
           </execution>
           <execution>
@@ -316,4 +317,16 @@
       </plugin>
     </plugins>
   </build>
+  <profiles>
+    <profile>
+      <id>jdk 13+</id>
+      <activation>
+        <jdk>[13,)</jdk>
+      </activation>
+      <properties>
+        <!-- for DriverBlockHoundIntegrationIT when using JDK 13+, see https://github.com/reactor/BlockHound/issues/33 -->
+        <blockhound.argline>-XX:+AllowRedefinitionToAddDeleteMethods</blockhound.argline>
+      </properties>
+    </profile>
+  </profiles>
 </project>

integration-tests/src/test/java/com/datastax/dse/driver/api/core/cql/continuous/ContinuousPagingIT.java

@@ -43,6 +43,7 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.Objects;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
@@ -183,7 +184,10 @@ public void simple_statement_paging_should_be_resilient_to_schema_change() {
             .getDefaultProfile()
             .withInt(DseDriverOption.CONTINUOUS_PAGING_MAX_ENQUEUED_PAGES, 1)
             .withInt(DseDriverOption.CONTINUOUS_PAGING_PAGE_SIZE, 1)
-            .withInt(DefaultDriverOption.REQUEST_TIMEOUT, 120000000);
+            .withDuration(
+                DseDriverOption.CONTINUOUS_PAGING_TIMEOUT_FIRST_PAGE, Duration.ofSeconds(30))
+            .withDuration(
+                DseDriverOption.CONTINUOUS_PAGING_TIMEOUT_OTHER_PAGES, Duration.ofSeconds(30));
     ContinuousResultSet result = session.executeContinuously(simple.setExecutionProfile(profile));
     Iterator<Row> it = result.iterator();
     // First row should have a non-null values.
@@ -193,11 +197,7 @@ public void simple_statement_paging_should_be_resilient_to_schema_change() {
     // Make schema change to add b, its metadata should NOT be present in subsequent rows.
     CqlSession schemaChangeSession =
         SessionUtils.newSession(
-            ccmRule,
-            session.getKeyspace().orElseThrow(IllegalStateException::new),
-            SessionUtils.configLoaderBuilder()
-                .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(30))
-                .build());
+            ccmRule, session.getKeyspace().orElseThrow(IllegalStateException::new));
     SimpleStatement statement =
         SimpleStatement.newInstance("ALTER TABLE test_prepare add b int")
             .setExecutionProfile(sessionRule.slowProfile());
@@ -251,7 +251,11 @@ public void prepared_statement_paging_should_be_resilient_to_schema_change() {
             .getConfig()
             .getDefaultProfile()
             .withInt(DseDriverOption.CONTINUOUS_PAGING_MAX_ENQUEUED_PAGES, 1)
-            .withInt(DseDriverOption.CONTINUOUS_PAGING_PAGE_SIZE, 1);
+            .withInt(DseDriverOption.CONTINUOUS_PAGING_PAGE_SIZE, 1)
+            .withDuration(
+                DseDriverOption.CONTINUOUS_PAGING_TIMEOUT_FIRST_PAGE, Duration.ofSeconds(30))
+            .withDuration(
+                DseDriverOption.CONTINUOUS_PAGING_TIMEOUT_OTHER_PAGES, Duration.ofSeconds(30));
     ContinuousResultSet result =
         session.executeContinuously(prepared.bind("foo").setExecutionProfile(profile));
     Iterator<Row> it = result.iterator();
@@ -262,11 +266,7 @@ public void prepared_statement_paging_should_be_resilient_to_schema_change() {
     // Make schema change to drop v, its metadata should be present, values will be null.
     CqlSession schemaChangeSession =
         SessionUtils.newSession(
-            ccmRule,
-            session.getKeyspace().orElseThrow(IllegalStateException::new),
-            SessionUtils.configLoaderBuilder()
-                .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, Duration.ofSeconds(30))
-                .build());
+            ccmRule, session.getKeyspace().orElseThrow(IllegalStateException::new));
     schemaChangeSession.execute("ALTER TABLE test_prep DROP v;");
     while (it.hasNext()) {
       // Each row should have a value for k, v should still be present, but null since column was
@@ -276,7 +276,7 @@ public void prepared_statement_paging_should_be_resilient_to_schema_change() {
       if (ccmRule
               .getDseVersion()
               .orElseThrow(IllegalStateException::new)
-              .compareTo(Version.parse("6.0.0"))
+              .compareTo(Objects.requireNonNull(Version.parse("6.0.0")))
           >= 0) {
         // DSE 6 only, v should be null here since dropped.
         // Not reliable for 5.1 since we may have gotten page queued before schema changed.

integration-tests/src/test/java/com/datastax/dse/driver/api/core/cql/continuous/ContinuousPagingITBase.java

@@ -105,17 +105,21 @@ public static Object[][] pagingOptions() {
 
   protected void validateMetrics(CqlSession session) {
     Node node = session.getMetadata().getNodes().values().iterator().next();
-    assertThat(session.getMetrics()).isPresent();
+    assertThat(session.getMetrics()).as("assert session.getMetrics() present").isPresent();
     Metrics metrics = session.getMetrics().get();
-    assertThat(metrics.getNodeMetric(node, DefaultNodeMetric.CQL_MESSAGES)).isPresent();
+    assertThat(metrics.getNodeMetric(node, DefaultNodeMetric.CQL_MESSAGES))
+        .as("assert metrics.getNodeMetric(node, DefaultNodeMetric.CQL_MESSAGES) present")
+        .isPresent();
     Timer messages = (Timer) metrics.getNodeMetric(node, DefaultNodeMetric.CQL_MESSAGES).get();
-    assertThat(messages.getCount()).isGreaterThan(0);
-    assertThat(messages.getMeanRate()).isGreaterThan(0);
-    assertThat(metrics.getSessionMetric(DseSessionMetric.CONTINUOUS_CQL_REQUESTS)).isPresent();
+    assertThat(messages.getCount()).as("assert messages.getCount() >= 0").isGreaterThan(0);
+    assertThat(messages.getMeanRate()).as("assert messages.getMeanRate() >= 0").isGreaterThan(0);
+    assertThat(metrics.getSessionMetric(DseSessionMetric.CONTINUOUS_CQL_REQUESTS))
+        .as("assert metrics.getSessionMetric(DseSessionMetric.CONTINUOUS_CQL_REQUESTS) present")
+        .isPresent();
     Timer requests =
         (Timer) metrics.getSessionMetric(DseSessionMetric.CONTINUOUS_CQL_REQUESTS).get();
-    assertThat(requests.getCount()).isGreaterThan(0);
-    assertThat(requests.getMeanRate()).isGreaterThan(0);
+    assertThat(requests.getCount()).as("assert requests.getCount() >= 0").isGreaterThan(0);
+    assertThat(requests.getMeanRate()).as("assert requests.getMeanRate() >= 0").isGreaterThan(0);
   }
 
   public static class Options {