3.x Driver may modify pipeline outside of event loop

Description

When negotiating a v5 connection, we re-configure the pipeline to handle the new framed format. This is intended to happen only on the event loop to ensure there are no races with processing received messages.
However, it appears that this is not always the case and we sometimes trigger an assertion error during initialisation. Stacktrace taken from a stress run:

{{java.lang.AssertionError
at com.datastax.driver.core.Connection.switchToV5Framing(Connection.java:1777)
at com.datastax.driver.core.Connection$6.apply(Connection.java:356)
at com.datastax.driver.core.Connection$6.apply(Connection.java:351)
at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:213)
at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:202)
at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:118)
at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1138)
at com.google.common.util.concurrent.AbstractFuture.addListener(AbstractFuture.java:707)
at com.google.common.util.concurrent.AbstractTransformFuture.create(AbstractTransformFuture.java:39)
at com.google.common.util.concurrent.Futures.transformAsync(Futures.java:464)
at com.datastax.driver.core.GuavaCompatibility$Version19OrHigher.transformAsync(GuavaCompatibility.java:280)
at com.datastax.driver.core.Connection$4.apply(Connection.java:321)
at com.datastax.driver.core.Connection$4.apply(Connection.java:313)
at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:213)
at com.google.common.util.concurrent.AbstractTransformFuture$AsyncTransformFuture.doTransform(AbstractTransformFuture.java:202)
at com.google.common.util.concurrent.AbstractTransformFuture.run(AbstractTransformFuture.java:118)
at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:30)
at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1138)
at com.google.common.util.concurrent.AbstractFuture.addListener(AbstractFuture.java:707)
at com.google.common.util.concurrent.AbstractFuture$TrustedFuture.addListener(AbstractFuture.java:112)
at com.google.common.util.concurrent.AbstractTransformFuture.create(AbstractTransformFuture.java:39)
at com.google.common.util.concurrent.Futures.transformAsync(Futures.java:464)
at com.datastax.driver.core.GuavaCompatibility$Version19OrHigher.transformAsync(GuavaCompatibility.java:280)
at com.datastax.driver.core.Connection.initAsync(Connection.java:246)
at com.datastax.driver.core.HostConnectionPool.initAsync(HostConnectionPool.java:149)
at com.datastax.driver.core.SessionManager.replacePool(SessionManager.java:360)
at com.datastax.driver.core.SessionManager.maybeAddPool(SessionManager.java:395)
at com.datastax.driver.core.SessionManager.createPools(SessionManager.java:127)
at com.datastax.driver.core.SessionManager.initAsync(SessionManager.java:94)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:389)
at com.datastax.driver.core.Cluster.connectAsync(Cluster.java:366)
at com.datastax.driver.core.Cluster.connect(Cluster.java:311)
at org.apache.cassandra.stress.util.JavaDriverClient.connect(JavaDriverClient.java:189)
at org.apache.cassandra.stress.settings.StressSettings.getJavaDriverClient(StressSettings.java:137)
at org.apache.cassandra.stress.settings.StressSettings.getJavaDriverClient(StressSettings.java:112)
at org.apache.cassandra.stress.settings.StressSettings.getJavaDriverClient(StressSettings.java:105)
at org.apache.cassandra.stress.StressProfile.maybeCreateSchema(StressProfile.java:224)
at org.apache.cassandra.stress.StressProfile.newGenerator(StressProfile.java:689)
at org.apache.cassandra.stress.StressProfile.printSettings(StressProfile.java:126)
at org.apache.cassandra.stress.settings.StressSettings.lambda$printSettings$1(StressSettings.java:309)
at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684)
at org.apache.cassandra.stress.settings.StressSettings.printSettings(StressSettings.java:309)
at org.apache.cassandra.stress.Stress.run(Stress.java:107)
at org.apache.cassandra.stress.Stress.main(Stress.java:63)
}}

Environment

None

Pull Requests

None
Fixed

Assignee

Alex Dutra

Reporter

Sam Tunnicliffe

Labels

None

PM Priority

None

Reproduced in

None

Affects versions

Fix versions

None

Pull Request

None

Doc Impact

None

Size

None

External issue ID

None

External issue ID

None

Components

Priority

Major