natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Avoid race condition when completing stream sessions 

patch by ZhaoYang; reviewed by Sergio Bossa and Benjamin Lerer for
CASSANDRA-15666
This commit is contained in:
Zhao Yang 2020-03-20 15:56:53 +08:00 committed by Benjamin Lerer
parent 9c1bbf3ac9
commit 9f6fcc340d
35 changed files with 675 additions and 285 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

174
.circleci/config.yml
View File

@ -3,7 +3,7 @@ jobs:
j8_jvm_upgrade_dtests:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -85,8 +85,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -94,7 +94,7 @@ jobs:
j8_cqlsh-dtests-py2-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -162,8 +162,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -171,7 +171,7 @@ jobs:
j11_unit_tests:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -253,8 +253,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -263,7 +263,7 @@ jobs:
j8_cqlsh-dtests-py38-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -331,8 +331,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -340,7 +340,7 @@ jobs:
j11_cqlsh-dtests-py3-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -408,8 +408,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -418,7 +418,7 @@ jobs:
j11_cqlsh-dtests-py3-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -486,8 +486,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -496,7 +496,7 @@ jobs:
j11_cqlsh-dtests-py38-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -564,8 +564,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -574,7 +574,7 @@ jobs:
j8_cqlsh-dtests-py3-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -642,8 +642,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -651,7 +651,7 @@ jobs:
j8_cqlsh-dtests-py2-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -719,8 +719,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -728,7 +728,7 @@ jobs:
j11_cqlsh-dtests-py2-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -796,8 +796,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -806,7 +806,7 @@ jobs:
j11_dtests-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -877,8 +877,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -887,7 +887,7 @@ jobs:
j8_dtests-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -936,8 +936,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -945,7 +945,7 @@ jobs:
j8_upgradetests-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1035,8 +1035,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1044,7 +1044,7 @@ jobs:
utests_stress:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1080,8 +1080,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1089,7 +1089,7 @@ jobs:
j8_unit_tests:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1171,8 +1171,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1180,7 +1180,7 @@ jobs:
j11_jvm_dtests:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1262,8 +1262,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -1272,7 +1272,7 @@ jobs:
j11_build:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1343,8 +1343,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -1353,7 +1353,7 @@ jobs:
j11_cqlsh-dtests-py2-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1421,8 +1421,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -1431,7 +1431,7 @@ jobs:
j8_dtests-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1480,8 +1480,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1489,7 +1489,7 @@ jobs:
j11_cqlsh-dtests-py38-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1557,8 +1557,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -1567,7 +1567,7 @@ jobs:
j8_jvm_dtests:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1649,8 +1649,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1658,7 +1658,7 @@ jobs:
j8_build:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1729,8 +1729,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1738,7 +1738,7 @@ jobs:
j8_cqlsh-dtests-py3-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1806,8 +1806,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1815,7 +1815,7 @@ jobs:
j8_cqlsh-dtests-py38-with-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -1883,8 +1883,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1892,7 +1892,7 @@ jobs:
utests_long:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1928,8 +1928,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1937,7 +1937,7 @@ jobs:
utests_fqltool:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -1973,8 +1973,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -1982,7 +1982,7 @@ jobs:
j11_dtests-no-vnodes:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -2053,8 +2053,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-11-openjdk-amd64
@ -2063,7 +2063,7 @@ jobs:
utests_compression:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 4
@ -2145,8 +2145,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64
@ -2154,7 +2154,7 @@ jobs:
j8_dtest_jars_build:
docker:
- image: nastra/cassandra-testing-ubuntu1910-java11-w-dependencies:20200406
resource_class: medium
resource_class: xlarge
working_directory: ~/
shell: /bin/bash -eo pipefail -l
parallelism: 1
@ -2213,8 +2213,8 @@ jobs:
- CASS_DRIVER_NO_EXTENSIONS: true
- CASS_DRIVER_NO_CYTHON: true
- CASSANDRA_SKIP_SYNC: true
- DTEST_REPO: git://github.com/apache/cassandra-dtest.git
- DTEST_BRANCH: master
- DTEST_REPO: git://github.com/jasonstack/cassandra-dtest.git
- DTEST_BRANCH: CASSANDRA-15666
- CCM_MAX_HEAP_SIZE: 1024M
- CCM_HEAP_NEWSIZE: 256M
- JAVA_HOME: /usr/lib/jvm/java-8-openjdk-amd64

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0-alpha4
* Avoid race condition when completing stream sessions (CASSANDRA-15666)
* Flush with fast compressors by default (CASSANDRA-15379)
* Fix CqlInputFormat regression from the switch to system.size_estimates (CASSANDRA-15637)
* Allow sending Entire SSTables over SSL (CASSANDRA-15740)

2
src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamWriter.java
View File

@ -62,7 +62,7 @@ public class CassandraCompressedStreamWriter extends CassandraStreamWriter
long totalSize = totalSize();
logger.debug("[Stream #{}] Start streaming file {} to {}, repairedAt = {}, totalSize = {}", session.planId(),
sstable.getFilename(), session.peer, sstable.getSSTableMetadata().repairedAt, totalSize);
try (ChannelProxy fc = sstable.getDataChannel().sharedCopy())
try (ChannelProxy fc = sstable.getDataChannel().newChannel())
{
long progress = 0L;

2
src/java/org/apache/cassandra/db/streaming/CassandraStreamWriter.java
View File

@ -83,7 +83,7 @@ public class CassandraStreamWriter
sstable.getFilename(), session.peer, sstable.getSSTableMetadata().repairedAt, totalSize);
AsyncStreamingOutputPlus out = (AsyncStreamingOutputPlus) output;
try(ChannelProxy proxy = sstable.getDataChannel().sharedCopy();
try(ChannelProxy proxy = sstable.getDataChannel().newChannel();
ChecksumValidator validator = new File(sstable.descriptor.filenameFor(Component.CRC)).exists()
? DataIntegrityMetadata.checksumValidator(sstable.descriptor)
: null)

10
src/java/org/apache/cassandra/io/util/ChannelProxy.java
View File

@ -111,6 +111,16 @@ public final class ChannelProxy extends SharedCloseableImpl
}
}
/**
* {@link #sharedCopy()} can not be used if thread will be interruped, as the backing channel will be closed.
*
* @return a new channel instance
*/
public final ChannelProxy newChannel()
{
return new ChannelProxy(filePath);
}
public ChannelProxy sharedCopy()
{
return new ChannelProxy(this);

4
src/java/org/apache/cassandra/net/OutboundConnection.java
View File

@ -1561,8 +1561,8 @@ public class OutboundConnection
Established established = state.established();
Channel channel = established.channel;
OutboundConnectionSettings settings = established.settings;
return SocketFactory.channelId(settings.from, (InetSocketAddress) channel.remoteAddress(),
settings.to, (InetSocketAddress) channel.localAddress(),
return SocketFactory.channelId(settings.from, (InetSocketAddress) channel.localAddress(),
settings.to, (InetSocketAddress) channel.remoteAddress(),
type, channel.id().asShortText());
}

2
src/java/org/apache/cassandra/service/QueryState.java
View File

@ -68,7 +68,7 @@ public class QueryState
public long getTimestamp()
{
if (timestamp == Long.MIN_VALUE)
timestamp = clientState.getTimestamp();
timestamp = ClientState.getTimestamp();
return timestamp;
}

15
src/java/org/apache/cassandra/streaming/StreamCoordinator.java
View File

@ -42,17 +42,19 @@ public class StreamCoordinator
private final Map<InetAddressAndPort, HostStreamingData> peerSessions = new HashMap<>();
private final StreamOperation streamOperation;
private final int connectionsPerHost;
private final boolean follower;
private StreamConnectionFactory factory;
private Iterator<StreamSession> sessionsToConnect = null;
private final UUID pendingRepair;
private final PreviewKind previewKind;
public StreamCoordinator(StreamOperation streamOperation, int connectionsPerHost, StreamConnectionFactory factory,
boolean connectSequentially, UUID pendingRepair, PreviewKind previewKind)
boolean follower, boolean connectSequentially, UUID pendingRepair, PreviewKind previewKind)
{
this.streamOperation = streamOperation;
this.connectionsPerHost = connectionsPerHost;
this.factory = factory;
this.follower = follower;
this.connectSequentially = connectSequentially;
this.pendingRepair = pendingRepair;
this.previewKind = previewKind;
@ -86,9 +88,9 @@ public class StreamCoordinator
return results;
}
public boolean isReceiving()
public boolean isFollower()
{
return connectionsPerHost == 0;
return follower;
}
public void connect(StreamResultFuture future)
@ -272,8 +274,7 @@ public class StreamCoordinator
{
for (StreamSession session : streamSessions.values())
{
StreamSession.State state = session.state();
if (state != StreamSession.State.COMPLETE && state != StreamSession.State.FAILED)
if (!session.state().isFinalState())
return true;
}
return false;
@ -284,7 +285,7 @@ public class StreamCoordinator
// create
if (streamSessions.size() < connectionsPerHost)
{
StreamSession session = new StreamSession(streamOperation, peer, factory, streamSessions.size(),
StreamSession session = new StreamSession(streamOperation, peer, factory, isFollower(), streamSessions.size(),
pendingRepair, previewKind);
streamSessions.put(++lastReturned, session);
return session;
@ -317,7 +318,7 @@ public class StreamCoordinator
StreamSession session = streamSessions.get(id);
if (session == null)
{
session = new StreamSession(streamOperation, peer, factory, id, pendingRepair, previewKind);
session = new StreamSession(streamOperation, peer, factory, isFollower(), id, pendingRepair, previewKind);
streamSessions.put(id, session);
}
return session;

28
src/java/org/apache/cassandra/streaming/StreamManager.java
View File

@ -106,12 +106,12 @@ public class StreamManager implements StreamManagerMBean
* We manage them in two different maps to distinguish plan from initiated ones to
* receiving ones withing the same JVM.
*/
private final Map<UUID, StreamResultFuture> initiatedStreams = new NonBlockingHashMap<>();
private final Map<UUID, StreamResultFuture> receivingStreams = new NonBlockingHashMap<>();
private final Map<UUID, StreamResultFuture> initiatorStreams = new NonBlockingHashMap<>();
private final Map<UUID, StreamResultFuture> followerStreams = new NonBlockingHashMap<>();
public Set<CompositeData> getCurrentStreams()
{
return Sets.newHashSet(Iterables.transform(Iterables.concat(initiatedStreams.values(), receivingStreams.values()), new Function<StreamResultFuture, CompositeData>()
return Sets.newHashSet(Iterables.transform(Iterables.concat(initiatorStreams.values(), followerStreams.values()), new Function<StreamResultFuture, CompositeData>()
{
public CompositeData apply(StreamResultFuture input)
{
@ -120,7 +120,7 @@ public class StreamManager implements StreamManagerMBean
}));
}
public void register(final StreamResultFuture result)
public void registerInitiator(final StreamResultFuture result)
{
result.addEventListener(notifier);
// Make sure we remove the stream on completion (whether successful or not)
@ -128,14 +128,14 @@ public class StreamManager implements StreamManagerMBean
{
public void run()
{
initiatedStreams.remove(result.planId);
initiatorStreams.remove(result.planId);
}
}, MoreExecutors.directExecutor());
initiatedStreams.put(result.planId, result);
initiatorStreams.put(result.planId, result);
}
public StreamResultFuture registerReceiving(final StreamResultFuture result)
public StreamResultFuture registerFollower(final StreamResultFuture result)
{
result.addEventListener(notifier);
// Make sure we remove the stream on completion (whether successful or not)
@ -143,17 +143,17 @@ public class StreamManager implements StreamManagerMBean
{
public void run()
{
receivingStreams.remove(result.planId);
followerStreams.remove(result.planId);
}
}, MoreExecutors.directExecutor());
StreamResultFuture previous = receivingStreams.putIfAbsent(result.planId, result);
StreamResultFuture previous = followerStreams.putIfAbsent(result.planId, result);
return previous == null ? result : previous;
}
public StreamResultFuture getReceivingStream(UUID planId)
{
return receivingStreams.get(planId);
return followerStreams.get(planId);
}
public void addNotificationListener(NotificationListener listener, NotificationFilter filter, Object handback)
@ -178,11 +178,15 @@ public class StreamManager implements StreamManagerMBean
public StreamSession findSession(InetAddressAndPort peer, UUID planId, int sessionIndex)
{
StreamSession session = findSession(initiatedStreams, peer, planId, sessionIndex);
// Search follower session first, because in some tests, eg. StreamingTransferTest, both initiator session
// and follower session are listening to local host.
// TODO CASSANDRA-15665 it's more robust to add "isFollower" flag into {@link StreamMessageHeader} to distinguish
// initiator session and follower session.
StreamSession session = findSession(followerStreams, peer, planId, sessionIndex);
if (session != null)
return session;
return findSession(receivingStreams, peer, planId, sessionIndex);
return findSession(initiatorStreams, peer, planId, sessionIndex);
}
private StreamSession findSession(Map<UUID, StreamResultFuture> streams, InetAddressAndPort peer, UUID planId, int sessionIndex)

4
src/java/org/apache/cassandra/streaming/StreamPlan.java
View File

@ -64,7 +64,7 @@ public class StreamPlan
{
this.streamOperation = streamOperation;
this.coordinator = new StreamCoordinator(streamOperation, connectionsPerHost, new DefaultConnectionFactory(),
connectSequentially, pendingRepair, previewKind);
false, connectSequentially, pendingRepair, previewKind);
}
/**
@ -176,7 +176,7 @@ public class StreamPlan
*/
public StreamResultFuture execute()
{
return StreamResultFuture.init(planId, streamOperation, handlers, coordinator);
return StreamResultFuture.createInitiator(planId, streamOperation, handlers, coordinator);
}
/**

31
src/java/org/apache/cassandra/streaming/StreamResultFuture.java
View File

@ -68,19 +68,19 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
this.coordinator = coordinator;
// if there is no session to listen to, we immediately set result for returning
if (!coordinator.isReceiving() && !coordinator.hasActiveSessions())
if (!coordinator.isFollower() && !coordinator.hasActiveSessions())
set(getCurrentState());
}
private StreamResultFuture(UUID planId, StreamOperation streamOperation, UUID pendingRepair, PreviewKind previewKind)
{
this(planId, streamOperation, new StreamCoordinator(streamOperation, 0, new DefaultConnectionFactory(), false, pendingRepair, previewKind));
this(planId, streamOperation, new StreamCoordinator(streamOperation, 0, new DefaultConnectionFactory(), true, false, pendingRepair, previewKind));
}
public static StreamResultFuture init(UUID planId, StreamOperation streamOperation, Collection<StreamEventHandler> listeners,
StreamCoordinator coordinator)
public static StreamResultFuture createInitiator(UUID planId, StreamOperation streamOperation, Collection<StreamEventHandler> listeners,
StreamCoordinator coordinator)
{
StreamResultFuture future = createAndRegister(planId, streamOperation, coordinator);
StreamResultFuture future = createAndRegisterInitiator(planId, streamOperation, coordinator);
if (listeners != null)
{
for (StreamEventHandler listener : listeners)
@ -100,13 +100,13 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
return future;
}
public static synchronized StreamResultFuture initReceivingSide(int sessionIndex,
UUID planId,
StreamOperation streamOperation,
InetAddressAndPort from,
Channel channel,
UUID pendingRepair,
PreviewKind previewKind)
public static synchronized StreamResultFuture createFollower(int sessionIndex,
UUID planId,
StreamOperation streamOperation,
InetAddressAndPort from,
Channel channel,
UUID pendingRepair,
PreviewKind previewKind)
{
StreamResultFuture future = StreamManager.instance.getReceivingStream(planId);
if (future == null)
@ -117,7 +117,7 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
// The main reason we create a StreamResultFuture on the receiving side is for JMX exposure.
future = new StreamResultFuture(planId, streamOperation, pendingRepair, previewKind);
StreamManager.instance.registerReceiving(future);
StreamManager.instance.registerFollower(future);
}
future.attachConnection(from, sessionIndex, channel);
logger.info("[Stream #{}, ID#{}] Received streaming plan for {} from {} channel.remote {} channel.local {} channel.id {}",
@ -125,10 +125,10 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
return future;
}
private static StreamResultFuture createAndRegister(UUID planId, StreamOperation streamOperation, StreamCoordinator coordinator)
private static StreamResultFuture createAndRegisterInitiator(UUID planId, StreamOperation streamOperation, StreamCoordinator coordinator)
{
StreamResultFuture future = new StreamResultFuture(planId, streamOperation, coordinator);
StreamManager.instance.register(future);
StreamManager.instance.registerInitiator(future);
return future;
}
@ -141,7 +141,6 @@ public final class StreamResultFuture extends AbstractFuture<StreamState>
{
StreamSession session = coordinator.getOrCreateSessionById(from, sessionIndex);
session.init(this);
session.attach(channel);
}
public void addEventListener(StreamEventHandler listener)

356
src/java/org/apache/cassandra/streaming/StreamSession.java
View File

@ -17,18 +17,18 @@
*/
package org.apache.cassandra.streaming;
import java.io.EOFException;
import java.net.SocketTimeoutException;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.*;
import com.google.common.util.concurrent.Futures;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.locator.RangesAtEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -54,10 +54,10 @@ import static org.apache.cassandra.net.MessagingService.current_version;
/**
* Handles the streaming a one or more streams to and from a specific remote node.
*
*<p/>
* Both this node and the remote one will create a similar symmetrical {@link StreamSession}. A streaming
* session has the following life-cycle:
*
*<pre>
* 1. Session Initialization
*
* (a) A node (the initiator in the following) create a new {@link StreamSession},
@ -101,12 +101,15 @@ import static org.apache.cassandra.net.MessagingService.current_version;
*
* 4. Completion phase
*
* (a) When a node enters the completion phase, it sends a {@link CompleteMessage} to the peer, and then enter the
* {@link StreamSession.State#WAIT_COMPLETE} state. If it has already received a {@link CompleteMessage}
* from the peer, session is complete and is then closed ({@link #closeSession(State)}). Otherwise, the node
* switch to the {@link StreamSession.State#WAIT_COMPLETE} state and send a {@link CompleteMessage} to the other side.
* (a) When the initiator finishes streaming, it enters the {@link StreamSession.State#WAIT_COMPLETE} state, and waits
* for the follower to send a {@link CompleteMessage} once it finishes streaming too. Once the {@link CompleteMessage}
* is received, initiator sets its own state to {@link StreamSession.State#COMPLETE} and closes all channels attached
* to this session.
*
* </pre>
*
* In brief, the message passing looks like this (I for initiator, F for follwer):
* <pre>
* (session init)
* I: StreamInitMessage
* (session prepare)
@ -117,7 +120,8 @@ import static org.apache.cassandra.net.MessagingService.current_version;
* I: OutgoingStreamMessage
* F: ReceivedMessage
* (completion)
* I/F: CompleteMessage
* F: CompleteMessage
*</pre>
*
* All messages which derive from {@link StreamMessage} are sent by the standard internode messaging
* (via {@link org.apache.cassandra.net.MessagingService}, while the actual files themselves are sent by a special
@ -127,6 +131,9 @@ public class StreamSession implements IEndpointStateChangeSubscriber
{
private static final Logger logger = LoggerFactory.getLogger(StreamSession.class);
// for test purpose to record received message and state transition
public volatile static MessageStateSink sink = MessageStateSink.NONE;
private final StreamOperation streamOperation;
/**
@ -153,43 +160,77 @@ public class StreamSession implements IEndpointStateChangeSubscriber
final Map<String, Set<Range<Token>>> transferredRangesPerKeyspace = new HashMap<>();
private final boolean isFollower;
private final NettyStreamingMessageSender messageSender;
private final ConcurrentMap<ChannelId, Channel> incomingChannels = new ConcurrentHashMap<>();
// contains both inbound and outbound channels
private final ConcurrentMap<ChannelId, Channel> channels = new ConcurrentHashMap<>();
// "maybeCompleted()" should be executed at most once. Because it can be executed asynchronously by IO
// threads(serialization/deserialization) and stream messaging processing thread, causing connection closed before
// receiving peer's CompleteMessage.
private boolean maybeCompleted = false;
private Future closeFuture;
private final AtomicBoolean isAborted = new AtomicBoolean(false);
private final UUID pendingRepair;
private final PreviewKind previewKind;
/**
* State Transition:
*
* <pre>
* +------------------+----------> FAILED <--------------------+
* | | ^ |
* | | | initiator |
* INITIALIZED --> PREPARING --> STREAMING ------------> WAIT_COMPLETE ----> COMPLETED
* | | | ^ ^
* | | | follower | |
* | | +-------------------------)-----------------+
* | | | |
* | | if preview | |
* | +----------------------------------------+ |
* | nothing to request or to transfer |
* +-----------------------------------------------------------------------------+
* nothing to request or to transfer
*
* </pre>
*/
public enum State
{
INITIALIZED,
PREPARING,
STREAMING,
WAIT_COMPLETE,
COMPLETE,
FAILED,
INITIALIZED(false),
PREPARING(false),
STREAMING(false),
WAIT_COMPLETE(false),
COMPLETE(true),
FAILED(true);
private final boolean finalState;
State(boolean finalState)
{
this.finalState = finalState;
}
/**
* @return true if current state is final, either COMPLETE OR FAILED.
*/
public boolean isFinalState()
{
return finalState;
}
}
private volatile State state = State.INITIALIZED;
private volatile boolean completeSent = false;
/**
* Create new streaming session with the peer.
*/
public StreamSession(StreamOperation streamOperation, InetAddressAndPort peer, StreamConnectionFactory factory,
int index, UUID pendingRepair, PreviewKind previewKind)
{
this(streamOperation, new OutboundConnectionSettings(peer), factory, index, pendingRepair, previewKind);
}
/**
* Create new streaming session with the peer.
*/
public StreamSession(StreamOperation streamOperation, OutboundConnectionSettings template, StreamConnectionFactory factory,
int index, UUID pendingRepair, PreviewKind previewKind)
boolean isFollower, int index, UUID pendingRepair, PreviewKind previewKind)
{
this.streamOperation = streamOperation;
this.peer = template.to;
this.template = template;
this.peer = peer;
this.template = new OutboundConnectionSettings(peer);
this.isFollower = isFollower;
this.index = index;
this.messageSender = new NettyStreamingMessageSender(this, template, factory, current_version, previewKind.isPreview());
@ -197,7 +238,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
this.pendingRepair = pendingRepair;
this.previewKind = previewKind;
logger.debug("Creating stream session to {}", template);
logger.debug("Creating stream session to {} as {}", template, isFollower ? "follower" : "initiator");
}
public UUID planId()
@ -253,11 +294,46 @@ public class StreamSession implements IEndpointStateChangeSubscriber
StreamHook.instance.reportStreamFuture(this, streamResult);
}
public boolean attach(Channel channel)
/**
* Attach a channel to this session upon receiving the first inbound message.
*
* @param channel The channel to attach.
* @param isControlChannel If the channel is the one to send control messages to.
* @return False if the channel was already attached, true otherwise.
*/
public synchronized boolean attachInbound(Channel channel, boolean isControlChannel)
{
if (!messageSender.hasControlChannel())
failIfFinished();
if (!messageSender.hasControlChannel() && isControlChannel)
messageSender.injectControlMessageChannel(channel);
return incomingChannels.putIfAbsent(channel.id(), channel) == null;
channel.closeFuture().addListener(ignored -> onChannelClose(channel));
return channels.putIfAbsent(channel.id(), channel) == null;
}
/**
* Attach a channel to this session upon sending the first outbound message.
*
* @param channel The channel to attach.
* @return False if the channel was already attached, true otherwise.
*/
public synchronized boolean attachOutbound(Channel channel)
{
failIfFinished();
channel.closeFuture().addListener(ignored -> onChannelClose(channel));
return channels.putIfAbsent(channel.id(), channel) == null;
}
/**
* On channel closing, if no channels are left just close the message sender; this must be closed last to ensure
* keep alive messages are sent until the very end of the streaming session.
*/
private synchronized void onChannelClose(Channel channel)
{
if (channels.remove(channel.id()) != null && channels.isEmpty())
messageSender.close();
}
/**
@ -339,7 +415,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
private void failIfFinished()
{
if (state() == State.COMPLETE || state() == State.FAILED)
if (state().isFinalState())
throw new RuntimeException(String.format("Stream %s is finished with state %s", planId(), state().name()));
}
@ -399,22 +475,32 @@ public class StreamSession implements IEndpointStateChangeSubscriber
private synchronized Future closeSession(State finalState)
{
Future abortedTasksFuture = null;
if (isAborted.compareAndSet(false, true))
// it's session is already closed
if (closeFuture != null)
return closeFuture;
state(finalState);
List<Future> futures = new ArrayList<>();
// ensure aborting the tasks do not happen on the network IO thread (read: netty event loop)
// as we don't want any blocking disk IO to stop the network thread
if (finalState == State.FAILED)
futures.add(ScheduledExecutors.nonPeriodicTasks.submit(this::abortTasks));
// Channels should only be closed by the initiator; but, if this session closed
// due to failure, channels should be always closed regardless, even if this is not the initator.
if (!isFollower || state != State.COMPLETE)
{
state(finalState);
// ensure aborting the tasks do not happen on the network IO thread (read: netty event loop)
// as we don't want any blocking disk IO to stop the network thread
if (finalState == State.FAILED)
abortedTasksFuture = ScheduledExecutors.nonPeriodicTasks.submit(this::abortTasks);
incomingChannels.values().stream().map(channel -> channel.close());
messageSender.close();
streamResult.handleSessionComplete(this);
logger.debug("[Stream #{}] Will close attached channels {}", planId(), channels);
channels.values().forEach(channel -> futures.add(channel.close()));
}
return abortedTasksFuture != null ? abortedTasksFuture : Futures.immediateFuture(null);
sink.onClose(peer);
streamResult.handleSessionComplete(this);
closeFuture = FBUtilities.allOf(futures);
return closeFuture;
}
private void abortTasks()
@ -426,7 +512,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
catch (Exception e)
{
logger.warn("failed to abort some streaming tasks", e);
logger.warn("[Stream #{}] failed to abort some streaming tasks", planId(), e);
}
}
@ -437,6 +523,10 @@ public class StreamSession implements IEndpointStateChangeSubscriber
*/
public void state(State newState)
{
if (logger.isTraceEnabled())
logger.trace("[Stream #{}] Changing session state from {} to {}", planId(), state, newState);
sink.recordState(peer, newState);
state = newState;
}
@ -463,21 +553,29 @@ public class StreamSession implements IEndpointStateChangeSubscriber
return state == State.COMPLETE;
}
public void messageReceived(StreamMessage message)
public synchronized void messageReceived(StreamMessage message)
{
if (message.type != StreamMessage.Type.KEEP_ALIVE)
failIfFinished();
sink.recordMessage(peer, message.type);
switch (message.type)
{
case STREAM_INIT:
// nop
// at follower, nop
break;
case PREPARE_SYN:
// at follower
PrepareSynMessage msg = (PrepareSynMessage) message;
prepare(msg.requests, msg.summaries);
break;
case PREPARE_SYNACK:
// at initiator
prepareSynAck((PrepareSynAckMessage) message);
break;
case PREPARE_ACK:
// at follower
prepareAck((PrepareAckMessage) message);
break;
case STREAM:
@ -488,6 +586,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
received(received.tableId, received.sequenceNumber);
break;
case COMPLETE:
// at initiator
complete();
break;
case KEEP_ALIVE:
@ -529,10 +628,32 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
/**
* Call back for handling exception during streaming.
* Signal an error to this stream session: if it's an EOF exception, it tries to understand if the socket was closed
* after completion or because the peer was down, otherwise sends a {@link SessionFailedMessage} and closes
* the session as {@link State#FAILED}.
*/
public Future onError(Throwable e)
public synchronized Future onError(Throwable e)
{
boolean isEofException = e instanceof EOFException;
if (isEofException)
{
if (state.finalState)
{
logger.debug("[Stream #{}] Socket closed after session completed with state {}", planId(), state);
return null;
}
else
{
logger.error("[Stream #{}] Socket closed before session completion, peer {} is probably down.",
planId(),
peer.address.getHostAddress(),
e);
return closeSession(State.FAILED);
}
}
logError(e);
// send session failure message
if (messageSender.connected())
@ -577,7 +698,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber
*/
private void prepareAsync(Collection<StreamRequest> requests, Collection<StreamSummary> summaries)
{
for (StreamRequest request : requests)
addTransferRanges(request.keyspace, RangesAtEndpoint.concat(request.full, request.transientReplicas), request.columnFamilies, true); // always flush on stream request
for (StreamSummary summary : summaries)
@ -589,9 +709,12 @@ public class StreamSession implements IEndpointStateChangeSubscriber
prepareSynAck.summaries.add(task.getSummary());
messageSender.sendMessage(prepareSynAck);
streamResult.handleSessionPrepared(this);
maybeCompleted();
if (isPreview())
completePreview();
else
maybeCompleted();
}
private void prepareSynAck(PrepareSynAckMessage msg)
@ -602,7 +725,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber
prepareReceiving(summary);
// only send the (final) ACK if we are expecting the peer to send this node (the initiator) some files
messageSender.sendMessage(new PrepareAckMessage());
if (!isPreview())
messageSender.sendMessage(new PrepareAckMessage());
}
if (isPreview())
@ -614,9 +738,8 @@ public class StreamSession implements IEndpointStateChangeSubscriber
private void prepareAck(PrepareAckMessage msg)
{
if (isPreview())
completePreview();
else
startStreamingFiles(true);
throw new RuntimeException(String.format("[Stream #%s] Cannot receive PrepareAckMessage for preview session", planId()));
startStreamingFiles(true);
}
/**
@ -646,7 +769,7 @@ public class StreamSession implements IEndpointStateChangeSubscriber
{
if (isPreview())
{
throw new RuntimeException("Cannot receive files for preview session");
throw new RuntimeException(String.format("[Stream #%s] Cannot receive files for preview session", planId()));
}
long headerSize = message.stream.getSize();
@ -674,22 +797,51 @@ public class StreamSession implements IEndpointStateChangeSubscriber
*/
public synchronized void complete()
{
logger.debug("handling Complete message, state = {}, completeSent = {}", state, completeSent);
if (state == State.WAIT_COMPLETE)
logger.debug("[Stream #{}] handling Complete message, state = {}", planId(), state);
if (!isFollower)
{
if (!completeSent)
{
messageSender.sendMessage(new CompleteMessage());
completeSent = true;
}
closeSession(State.COMPLETE);
if (state == State.WAIT_COMPLETE)
closeSession(State.COMPLETE);
else
state(State.WAIT_COMPLETE);
}
else
{
state(State.WAIT_COMPLETE);
// pre-4.0 nodes should not be connected via streaming, see {@link MessagingService#accept_streaming}
throw new IllegalStateException(String.format("[Stream #%s] Complete message can be only received by the initiator!", planId()));
}
}
/**
* Synchronize both {@link #complete()} and {@link #maybeCompleted()} to avoid racing
*/
private synchronized boolean maybeCompleted()
{
if (!(receivers.isEmpty() && transfers.isEmpty()))
return false;
// if already executed once, skip it
if (maybeCompleted)
return true;
maybeCompleted = true;
if (!isFollower)
{
if (state == State.WAIT_COMPLETE)
closeSession(State.COMPLETE);
else
state(State.WAIT_COMPLETE);
}
else
{
messageSender.sendMessage(new CompleteMessage());
closeSession(State.COMPLETE);
}
return true;
}
/**
* Call back on receiving {@code StreamMessage.Type.SESSION_FAILED} message.
*/
@ -760,31 +912,6 @@ public class StreamSession implements IEndpointStateChangeSubscriber
}
}
private boolean maybeCompleted()
{
boolean completed = receivers.isEmpty() && transfers.isEmpty();
if (completed)
{
if (state == State.WAIT_COMPLETE)
{
if (!completeSent)
{
messageSender.sendMessage(new CompleteMessage());
completeSent = true;
}
closeSession(State.COMPLETE);
}
else
{
// notify peer that this session is completed
messageSender.sendMessage(new CompleteMessage());
completeSent = true;
state(State.WAIT_COMPLETE);
}
}
return completed;
}
/**
* Flushes matching column families from the given keyspace, or all columnFamilies
* if the cf list is empty.
@ -843,4 +970,43 @@ public class StreamSession implements IEndpointStateChangeSubscriber
{
return transferredRangesPerKeyspace.size();
}
@VisibleForTesting
public static interface MessageStateSink
{
static final MessageStateSink NONE = new MessageStateSink() {
@Override
public void recordState(InetAddressAndPort from, State state)
{
}
@Override
public void recordMessage(InetAddressAndPort from, StreamMessage.Type message)
{
}
@Override
public void onClose(InetAddressAndPort from)
{
}
};
/**
* @param from peer that is connected in the stream session
* @param state new state to change to
*/
public void recordState(InetAddressAndPort from, StreamSession.State state);
/**
* @param from peer that sends the given message
* @param message stream message sent by peer
*/
public void recordMessage(InetAddressAndPort from, StreamMessage.Type message);
/**
*
* @param from peer that is being disconnected
*/
public void onClose(InetAddressAndPort from);
}
}

60
src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java
View File

@ -21,9 +21,8 @@ package org.apache.cassandra.streaming.async;
import java.io.IOError;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.nio.channels.ClosedByInterruptException;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
@ -33,6 +32,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -61,6 +61,7 @@ import org.apache.cassandra.streaming.messages.OutgoingStreamMessage;
import org.apache.cassandra.streaming.messages.StreamInitMessage;
import org.apache.cassandra.streaming.messages.StreamMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
/**
* Responsible for sending {@link StreamMessage}s to a given peer. We manage an array of netty {@link Channel}s
@ -102,7 +103,7 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
* A special {@link Channel} for sending non-stream streaming messages, basically anything that isn't an
* {@link OutgoingStreamMessage} (or an {@link IncomingStreamMessage}, but a node doesn't send that, it's only received).
*/
private Channel controlMessageChannel;
private volatile Channel controlMessageChannel;
// note: this really doesn't need to be a LBQ, just something that's thread safe
private final Collection<ScheduledFuture<?>> channelKeepAlives = new LinkedBlockingQueue<>();
@ -153,6 +154,9 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
return controlMessageChannel != null;
}
/**
* Used by follower to setup control message channel created by initiator
*/
public void injectControlMessageChannel(Channel channel)
{
this.controlMessageChannel = channel;
@ -160,11 +164,20 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
scheduleKeepAliveTask(channel);
}
/**
* Used by initiator to setup control message channel connecting to follower
*/
private void setupControlMessageChannel() throws IOException
{
if (controlMessageChannel == null)
{
controlMessageChannel = createChannel();
/*
* Inbound handlers are needed:
* a) for initiator's control channel(the first outbound channel) to receive follower's message.
* b) for streaming receiver (note: both initiator and follower can receive streaming files) to reveive files,
* in {@link Handler#setupStreamingPipeline}
*/
controlMessageChannel = createChannel(true);
scheduleKeepAliveTask(controlMessageChannel);
}
}
@ -181,11 +194,16 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
task.future = scheduledFuture;
}
private Channel createChannel() throws IOException
private Channel createChannel(boolean isInboundHandlerNeeded) throws IOException
{
Channel channel = factory.createConnection(template, streamingVersion);
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast("stream", new StreamingInboundHandler(template.to, streamingVersion, session));
session.attachOutbound(channel);
if (isInboundHandlerNeeded)
{
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast("stream", new StreamingInboundHandler(template.to, streamingVersion, session));
}
channel.attr(TRANSFERRING_FILE_ATTR).set(Boolean.FALSE);
logger.debug("Creating channel id {} local {} remote {}", channel.id(), channel.localAddress(), channel.remoteAddress());
return channel;
@ -316,9 +334,10 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
if (!acquirePermit(SEMAPHORE_UNAVAILABLE_LOG_INTERVAL))
return;
Channel channel = null;
try
{
Channel channel = getOrCreateChannel();
channel = getOrCreateChannel();
if (!channel.attr(TRANSFERRING_FILE_ATTR).compareAndSet(false, true))
throw new IllegalStateException("channel's transferring state is currently set to true. refusing to start new stream");
@ -336,6 +355,19 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
{
session.onError(e);
}
catch (Throwable t)
{
if (closed && Throwables.getRootCause(t) instanceof ClosedByInterruptException && fileTransferExecutor.isShutdown())
{
logger.debug("{} Streaming channel was closed due to the executor pool being shutdown", createLogTag(session, channel));
}
else
{
JVMStabilityInspector.inspectThrowable(t);
if (!session.state().isFinalState())
session.onError(t);
}
}
finally
{
fileTransferSemaphore.release();
@ -383,7 +415,7 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
if (channel != null)
return channel;
channel = createChannel();
channel = createChannel(false);
threadToChannelMap.put(currentThread, channel);
return channel;
}
@ -513,6 +545,9 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
@Override
public void close()
{
if (closed)
return;
closed = true;
if (logger.isDebugEnabled())
logger.debug("{} Closing stream connection channels on {}", createLogTag(session, null), template.to);
@ -520,14 +555,7 @@ public class NettyStreamingMessageSender implements StreamingMessageSender
future.cancel(false);
channelKeepAlives.clear();
List<Future<Void>> futures = new ArrayList<>(threadToChannelMap.size());
for (Channel channel : threadToChannelMap.values())
futures.add(channel.close());
FBUtilities.waitOnFutures(futures, 10, TimeUnit.SECONDS);
threadToChannelMap.clear();
fileTransferExecutor.shutdownNow();
if (controlMessageChannel != null)
controlMessageChannel.close();
}
}

17
src/java/org/apache/cassandra/streaming/async/StreamingInboundHandler.java
View File

@ -18,12 +18,9 @@
package org.apache.cassandra.streaming.async;
import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@ -31,7 +28,6 @@ import java.util.function.Function;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -44,7 +40,6 @@ import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.FastThreadLocalThread;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.AsyncStreamingInputPlus;
import org.apache.cassandra.net.AsyncStreamingInputPlus.InputTimeoutException;
import org.apache.cassandra.streaming.StreamManager;
import org.apache.cassandra.streaming.StreamReceiveException;
import org.apache.cassandra.streaming.StreamResultFuture;
@ -183,7 +178,7 @@ public class StreamingInboundHandler extends ChannelInboundHandlerAdapter
Uninterruptibles.sleepUninterruptibly(400, TimeUnit.MILLISECONDS);
}
StreamMessage message = StreamMessage.deserialize(buffers, protocolVersion, null);
StreamMessage message = StreamMessage.deserialize(buffers, protocolVersion);
// keep-alives don't necessarily need to be tied to a session (they could be arrive before or after
// wrt session lifecycle, due to races), just log that we received the message and carry on
@ -203,10 +198,6 @@ public class StreamingInboundHandler extends ChannelInboundHandlerAdapter
session.messageReceived(message);
}
}
catch (InputTimeoutException | EOFException e)
{
// ignore
}
catch (Throwable t)
{
JVMStabilityInspector.inspectThrowable(t);
@ -248,7 +239,7 @@ public class StreamingInboundHandler extends ChannelInboundHandlerAdapter
{
assert session == null : "initiator of stream session received a StreamInitMessage";
StreamInitMessage init = (StreamInitMessage) message;
StreamResultFuture.initReceivingSide(init.sessionIndex, init.planId, init.streamOperation, init.from, channel, init.pendingRepair, init.previewKind);
StreamResultFuture.createFollower(init.sessionIndex, init.planId, init.streamOperation, init.from, channel, init.pendingRepair, init.previewKind);
streamSession = sessionProvider.apply(new SessionIdentifier(init.from, init.planId, init.sessionIndex));
}
else if (message instanceof IncomingStreamMessage)
@ -262,7 +253,9 @@ public class StreamingInboundHandler extends ChannelInboundHandlerAdapter
if (streamSession == null)
throw new IllegalStateException(createLogTag(null, channel) + " no session found for message " + message);
streamSession.attach(channel);
// Attach this channel to the session: this only happens upon receiving the first init message as a follower;
// in all other cases, no new control channel will be added, as the proper control channel will be already attached.
streamSession.attachInbound(channel, message instanceof StreamInitMessage);
return streamSession;
}
}

2
src/java/org/apache/cassandra/streaming/messages/CompleteMessage.java
View File

@ -25,7 +25,7 @@ public class CompleteMessage extends StreamMessage
{
public static Serializer<CompleteMessage> serializer = new Serializer<CompleteMessage>()
{
public CompleteMessage deserialize(DataInputPlus in, int version, StreamSession session)
public CompleteMessage deserialize(DataInputPlus in, int version)
{
return new CompleteMessage();
}

4
src/java/org/apache/cassandra/streaming/messages/IncomingStreamMessage.java
View File

@ -35,10 +35,10 @@ public class IncomingStreamMessage extends StreamMessage
public static Serializer<IncomingStreamMessage> serializer = new Serializer<IncomingStreamMessage>()
{
@SuppressWarnings("resource")
public IncomingStreamMessage deserialize(DataInputPlus input, int version, StreamSession session) throws IOException
public IncomingStreamMessage deserialize(DataInputPlus input, int version) throws IOException
{
StreamMessageHeader header = StreamMessageHeader.serializer.deserialize(input, version);
session = StreamManager.instance.findSession(header.sender, header.planId, header.sessionIndex);
StreamSession session = StreamManager.instance.findSession(header.sender, header.planId, header.sessionIndex);
if (session == null)
throw new IllegalStateException(String.format("unknown stream session: %s - %d", header.planId, header.sessionIndex));
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(header.tableId);

2
src/java/org/apache/cassandra/streaming/messages/KeepAliveMessage.java
View File

@ -28,7 +28,7 @@ public class KeepAliveMessage extends StreamMessage
{
public static Serializer<KeepAliveMessage> serializer = new Serializer<KeepAliveMessage>()
{
public KeepAliveMessage deserialize(DataInputPlus in, int version, StreamSession session) throws IOException
public KeepAliveMessage deserialize(DataInputPlus in, int version) throws IOException
{
return new KeepAliveMessage();
}

2
src/java/org/apache/cassandra/streaming/messages/OutgoingStreamMessage.java
View File

@ -32,7 +32,7 @@ public class OutgoingStreamMessage extends StreamMessage
{
public static Serializer<OutgoingStreamMessage> serializer = new Serializer<OutgoingStreamMessage>()
{
public OutgoingStreamMessage deserialize(DataInputPlus in, int version, StreamSession session)
public OutgoingStreamMessage deserialize(DataInputPlus in, int version)
{
throw new UnsupportedOperationException("Not allowed to call deserialize on an outgoing stream");
}

2
src/java/org/apache/cassandra/streaming/messages/PrepareAckMessage.java
View File

@ -33,7 +33,7 @@ public class PrepareAckMessage extends StreamMessage
//nop
}
public PrepareAckMessage deserialize(DataInputPlus in, int version, StreamSession session) throws IOException
public PrepareAckMessage deserialize(DataInputPlus in, int version) throws IOException
{
return new PrepareAckMessage();
}

2
src/java/org/apache/cassandra/streaming/messages/PrepareSynAckMessage.java
View File

@ -38,7 +38,7 @@ public class PrepareSynAckMessage extends StreamMessage
StreamSummary.serializer.serialize(summary, out, version);
}
public PrepareSynAckMessage deserialize(DataInputPlus input, int version, StreamSession session) throws IOException
public PrepareSynAckMessage deserialize(DataInputPlus input, int version) throws IOException
{
PrepareSynAckMessage message = new PrepareSynAckMessage();
int numSummaries = input.readInt();

2
src/java/org/apache/cassandra/streaming/messages/PrepareSynMessage.java
View File

@ -31,7 +31,7 @@ public class PrepareSynMessage extends StreamMessage
{
public static Serializer<PrepareSynMessage> serializer = new Serializer<PrepareSynMessage>()
{
public PrepareSynMessage deserialize(DataInputPlus input, int version, StreamSession session) throws IOException
public PrepareSynMessage deserialize(DataInputPlus input, int version) throws IOException
{
PrepareSynMessage message = new PrepareSynMessage();
// requests

2
src/java/org/apache/cassandra/streaming/messages/ReceivedMessage.java
View File

@ -29,7 +29,7 @@ public class ReceivedMessage extends StreamMessage
public static Serializer<ReceivedMessage> serializer = new Serializer<ReceivedMessage>()
{
@SuppressWarnings("resource") // Not closing constructed DataInputPlus's as the channel needs to remain open.
public ReceivedMessage deserialize(DataInputPlus input, int version, StreamSession session) throws IOException
public ReceivedMessage deserialize(DataInputPlus input, int version) throws IOException
{
return new ReceivedMessage(TableId.deserialize(input), input.readInt());
}

2
src/java/org/apache/cassandra/streaming/messages/SessionFailedMessage.java
View File

@ -25,7 +25,7 @@ public class SessionFailedMessage extends StreamMessage
{
public static Serializer<SessionFailedMessage> serializer = new Serializer<SessionFailedMessage>()
{
public SessionFailedMessage deserialize(DataInputPlus in, int version, StreamSession session)
public SessionFailedMessage deserialize(DataInputPlus in, int version)
{
return new SessionFailedMessage();
}

2
src/java/org/apache/cassandra/streaming/messages/StreamInitMessage.java
View File

@ -86,7 +86,7 @@ public class StreamInitMessage extends StreamMessage
out.writeInt(message.previewKind.getSerializationVal());
}
public StreamInitMessage deserialize(DataInputPlus in, int version, StreamSession session) throws IOException
public StreamInitMessage deserialize(DataInputPlus in, int version) throws IOException
{
InetAddressAndPort from = inetAddressAndPortSerializer.deserialize(in, version);
int sessionIndex = in.readInt();

6
src/java/org/apache/cassandra/streaming/messages/StreamMessage.java
View File

@ -43,16 +43,16 @@ public abstract class StreamMessage
return 1 + message.type.outSerializer.serializedSize(message, version);
}
public static StreamMessage deserialize(DataInputPlus in, int version, StreamSession session) throws IOException
public static StreamMessage deserialize(DataInputPlus in, int version) throws IOException
{
Type type = Type.lookupById(in.readByte());
return type.inSerializer.deserialize(in, version, session);
return type.inSerializer.deserialize(in, version);
}
/** StreamMessage serializer */
public static interface Serializer<V extends StreamMessage>
{
V deserialize(DataInputPlus in, int version, StreamSession session) throws IOException;
V deserialize(DataInputPlus in, int version) throws IOException;
void serialize(V message, DataOutputStreamPlus out, int version, StreamSession session) throws IOException;
long serializedSize(V message, int version) throws IOException;
}

75
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -486,6 +486,81 @@ public class FBUtilities
Uninterruptibles.sleepUninterruptibly(delay, TimeUnit.MILLISECONDS);
}
}
/**
* Returns a new {@link Future} wrapping the given list of futures and returning a list of their results.
*/
public static Future<List> allOf(Collection<Future> futures)
{
if (futures.isEmpty())
return CompletableFuture.completedFuture(null);
return new Future<List>()
{
@Override
@SuppressWarnings("unchecked")
public List get() throws InterruptedException, ExecutionException
{
List result = new ArrayList<>(futures.size());
for (Future current : futures)
{
result.add(current.get());
}
return result;
}
@Override
@SuppressWarnings("unchecked")
public List get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
List result = new ArrayList<>(futures.size());
long deadline = System.nanoTime() + TimeUnit.NANOSECONDS.convert(timeout, unit);
for (Future current : futures)
{
long remaining = deadline - System.nanoTime();
if (remaining <= 0)
throw new TimeoutException();
result.add(current.get(remaining, TimeUnit.NANOSECONDS));
}
return result;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
for (Future current : futures)
{
if (!current.cancel(mayInterruptIfRunning))
return false;
}
return true;
}
@Override
public boolean isCancelled()
{
for (Future current : futures)
{
if (!current.isCancelled())
return false;
}
return true;
}
@Override
public boolean isDone()
{
for (Future current : futures)
{
if (!current.isDone())
return false;
}
return true;
}
};
}
/**
* Create a new instance of a partitioner defined in an SSTable Descriptor
* @param desc Descriptor of an sstable

107
test/distributed/org/apache/cassandra/distributed/test/StreamingTest.java
View File

@ -18,16 +18,39 @@
package org.apache.cassandra.distributed.test;
import java.io.Serializable;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
import org.junit.Assert;
import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.streaming.messages.StreamMessage;
import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.streaming.StreamSession.State.PREPARING;
import static org.apache.cassandra.streaming.StreamSession.State.STREAMING;
import static org.apache.cassandra.streaming.StreamSession.State.WAIT_COMPLETE;
import static org.apache.cassandra.streaming.messages.StreamMessage.Type.PREPARE_ACK;
import static org.apache.cassandra.streaming.messages.StreamMessage.Type.PREPARE_SYN;
import static org.apache.cassandra.streaming.messages.StreamMessage.Type.PREPARE_SYNACK;
import static org.apache.cassandra.streaming.messages.StreamMessage.Type.RECEIVED;
import static org.apache.cassandra.streaming.messages.StreamMessage.Type.STREAM;
import static org.apache.cassandra.streaming.messages.StreamMessage.Type.STREAM_INIT;
public class StreamingTest extends TestBaseImpl
{
@ -51,6 +74,9 @@ public class StreamingTest extends TestBaseImpl
Assert.assertEquals(0, results.length);
}
// collect message and state
registerSink(cluster, nodes);
cluster.get(nodes).runOnInstance(() -> StorageService.instance.rebuild(null, KEYSPACE, null, null));
{
Object[][] results = cluster.get(nodes).executeInternal(String.format("SELECT k, c1, c2 FROM %s.cf;", KEYSPACE));
@ -72,4 +98,85 @@ public class StreamingTest extends TestBaseImpl
testStreaming(2, 2, 1000, "LeveledCompactionStrategy");
}
public static void registerSink(Cluster cluster, int initiatorNodeId)
{
IInvokableInstance initiatorNode = cluster.get(initiatorNodeId);
InetSocketAddress initiator = initiatorNode.broadcastAddress();
MessageStateSinkImpl initiatorSink = new MessageStateSinkImpl();
for (int node = 1; node <= cluster.size(); node++)
{
if (initiatorNodeId == node)
continue;
IInvokableInstance followerNode = cluster.get(node);
InetSocketAddress follower = followerNode.broadcastAddress();
// verify on initiator's stream session
initiatorSink.messages(follower, Arrays.asList(PREPARE_SYNACK, STREAM, StreamMessage.Type.COMPLETE));
initiatorSink.states(follower, Arrays.asList(PREPARING, STREAMING, WAIT_COMPLETE, StreamSession.State.COMPLETE));
// verify on follower's stream session
MessageStateSinkImpl followerSink = new MessageStateSinkImpl();
followerSink.messages(initiator, Arrays.asList(STREAM_INIT, PREPARE_SYN, PREPARE_ACK, RECEIVED));
followerSink.states(initiator, Arrays.asList(PREPARING, STREAMING, StreamSession.State.COMPLETE));
followerNode.runOnInstance(() -> StreamSession.sink = followerSink);
}
cluster.get(initiatorNodeId).runOnInstance(() -> StreamSession.sink = initiatorSink);
}
@VisibleForTesting
public static class MessageStateSinkImpl implements StreamSession.MessageStateSink, Serializable
{
// use enum ordinal instead of enum to walk around inter-jvm class loader issue, only classes defined in
// InstanceClassLoader#sharedClassNames are shareable between server jvm and test jvm
public final Map<InetAddress, Queue<Integer>> messageSink = new ConcurrentHashMap<>();
public final Map<InetAddress, Queue<Integer>> stateTransitions = new ConcurrentHashMap<>();
public void messages(InetSocketAddress peer, List<StreamMessage.Type> messages)
{
messageSink.put(peer.getAddress(), messages.stream().map(Enum::ordinal).collect(Collectors.toCollection(LinkedList::new)));
}
public void states(InetSocketAddress peer, List<StreamSession.State> states)
{
stateTransitions.put(peer.getAddress(), states.stream().map(Enum::ordinal).collect(Collectors.toCollection(LinkedList::new)));
}
@Override
public void recordState(InetAddressAndPort from, StreamSession.State state)
{
Queue<Integer> states = stateTransitions.get(from.address);
if (states.peek() == null)
Assert.fail("Unexpected state " + state);
int expected = states.poll();
Assert.assertEquals(StreamSession.State.values()[expected], state);
}
@Override
public void recordMessage(InetAddressAndPort from, StreamMessage.Type message)
{
if (message == StreamMessage.Type.KEEP_ALIVE)
return;
Queue<Integer> messages = messageSink.get(from.address);
if (messages.peek() == null)
Assert.fail("Unexpected message " + message);
int expected = messages.poll();
Assert.assertEquals(StreamMessage.Type.values()[expected], message);
}
@Override
public void onClose(InetAddressAndPort from)
{
Queue<Integer> states = stateTransitions.get(from.address);
Assert.assertTrue("Missing states: " + states, states.isEmpty());
Queue<Integer> messages = messageSink.get(from.address);
Assert.assertTrue("Missing messages: " + messages, messages.isEmpty());
}
}
}

4
test/microbench/org/apache/cassandra/test/microbench/ZeroCopyStreamingBenchmark.java
View File

@ -213,8 +213,8 @@ public class ZeroCopyStreamingBenchmark
private StreamSession setupStreamingSessionForTest()
{
StreamCoordinator streamCoordinator = new StreamCoordinator(StreamOperation.BOOTSTRAP, 1, new DefaultConnectionFactory(), false, null, PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.init(UUID.randomUUID(), StreamOperation.BOOTSTRAP, Collections.<StreamEventHandler>emptyList(), streamCoordinator);
StreamCoordinator streamCoordinator = new StreamCoordinator(StreamOperation.BOOTSTRAP, 1, new DefaultConnectionFactory(), false, false, null, PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.createInitiator(UUID.randomUUID(), StreamOperation.BOOTSTRAP, Collections.<StreamEventHandler>emptyList(), streamCoordinator);
InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort();
streamCoordinator.addSessionInfo(new SessionInfo(peer, 0, peer, Collections.emptyList(), Collections.emptyList(), StreamSession.State.INITIALIZED));

4
test/unit/org/apache/cassandra/db/streaming/CassandraEntireSSTableStreamWriterTest.java
View File

@ -196,8 +196,8 @@ public class CassandraEntireSSTableStreamWriterTest
private StreamSession setupStreamingSessionForTest()
{
StreamCoordinator streamCoordinator = new StreamCoordinator(StreamOperation.BOOTSTRAP, 1, new DefaultConnectionFactory(), false, null, PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.init(UUID.randomUUID(), StreamOperation.BOOTSTRAP, Collections.<StreamEventHandler>emptyList(), streamCoordinator);
StreamCoordinator streamCoordinator = new StreamCoordinator(StreamOperation.BOOTSTRAP, 1, new DefaultConnectionFactory(), false, false, null, PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.createInitiator(UUID.randomUUID(), StreamOperation.BOOTSTRAP, Collections.<StreamEventHandler>emptyList(), streamCoordinator);
InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort();
streamCoordinator.addSessionInfo(new SessionInfo(peer, 0, peer, Collections.emptyList(), Collections.emptyList(), StreamSession.State.INITIALIZED));

1
test/unit/org/apache/cassandra/db/streaming/CassandraStreamManagerTest.java
View File

@ -99,6 +99,7 @@ public class CassandraStreamManagerTest
return new StreamSession(StreamOperation.REPAIR,
InetAddressAndPort.getByName("127.0.0.1"),
connectionFactory,
false,
0,
pendingRepair,
PreviewKind.NONE);

4
test/unit/org/apache/cassandra/dht/StreamStateStoreTest.java
View File

@ -55,7 +55,7 @@ public class StreamStateStoreTest
Range<Token> range = new Range<>(factory.fromString("0"), factory.fromString("100"));
InetAddressAndPort local = FBUtilities.getBroadcastAddressAndPort();
StreamSession session = new StreamSession(StreamOperation.BOOTSTRAP, local, new DefaultConnectionFactory(), 0, null, PreviewKind.NONE);
StreamSession session = new StreamSession(StreamOperation.BOOTSTRAP, local, new DefaultConnectionFactory(), false, 0, null, PreviewKind.NONE);
session.addStreamRequest("keyspace1", RangesAtEndpoint.toDummyList(Collections.singleton(range)), RangesAtEndpoint.toDummyList(Collections.emptyList()), Collections.singleton("cf"));
StreamStateStore store = new StreamStateStore();
@ -76,7 +76,7 @@ public class StreamStateStoreTest
// add different range within the same keyspace
Range<Token> range2 = new Range<>(factory.fromString("100"), factory.fromString("200"));
session = new StreamSession(StreamOperation.BOOTSTRAP, local, new DefaultConnectionFactory(), 0, null, PreviewKind.NONE);
session = new StreamSession(StreamOperation.BOOTSTRAP, local, new DefaultConnectionFactory(), false, 0, null, PreviewKind.NONE);
session.addStreamRequest("keyspace1", RangesAtEndpoint.toDummyList(Collections.singleton(range2)), RangesAtEndpoint.toDummyList(Collections.emptyList()), Collections.singleton("cf"));
session.state(StreamSession.State.COMPLETE);
store.handleStreamEvent(new StreamEvent.SessionCompleteEvent(session));

9
test/unit/org/apache/cassandra/streaming/EntireSSTableStreamingCorrectFilesCountTest.java
View File

@ -187,13 +187,14 @@ public class EntireSSTableStreamingCorrectFilesCountTest
1,
new DefaultConnectionFactory(),
false,
false,
null,
PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.init(UUID.randomUUID(),
StreamOperation.BOOTSTRAP,
Collections.singleton(streamEventHandler),
streamCoordinator);
StreamResultFuture future = StreamResultFuture.createInitiator(UUID.randomUUID(),
StreamOperation.BOOTSTRAP,
Collections.singleton(streamEventHandler),
streamCoordinator);
InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort();
streamCoordinator.addSessionInfo(new SessionInfo(peer,

10
test/unit/org/apache/cassandra/streaming/StreamTransferTaskTest.java
View File

@ -77,7 +77,7 @@ public class StreamTransferTaskTest
public void testScheduleTimeout() throws Exception
{
InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort();
StreamSession session = new StreamSession(StreamOperation.BOOTSTRAP, peer, (template, messagingVersion) -> new EmbeddedChannel(), 0, UUID.randomUUID(), PreviewKind.ALL);
StreamSession session = new StreamSession(StreamOperation.BOOTSTRAP, peer, (template, messagingVersion) -> new EmbeddedChannel(), false, 0, UUID.randomUUID(), PreviewKind.ALL);
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD);
// create two sstables
@ -88,6 +88,7 @@ public class StreamTransferTaskTest
}
// create streaming task that streams those two sstables
session.state(StreamSession.State.PREPARING);
StreamTransferTask task = new StreamTransferTask(session, cfs.metadata.id);
for (SSTableReader sstable : cfs.getLiveSSTables())
{
@ -98,6 +99,7 @@ public class StreamTransferTaskTest
assertEquals(14, task.getTotalNumberOfFiles());
// if file sending completes before timeout then the task should be canceled.
session.state(StreamSession.State.STREAMING);
Future f = task.scheduleTimeout(0, 0, TimeUnit.NANOSECONDS);
f.get();
@ -123,9 +125,9 @@ public class StreamTransferTaskTest
public void testFailSessionDuringTransferShouldNotReleaseReferences() throws Exception
{
InetAddressAndPort peer = FBUtilities.getBroadcastAddressAndPort();
StreamCoordinator streamCoordinator = new StreamCoordinator(StreamOperation.BOOTSTRAP, 1, new DefaultConnectionFactory(), false, null, PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.init(UUID.randomUUID(), StreamOperation.OTHER, Collections.<StreamEventHandler>emptyList(), streamCoordinator);
StreamSession session = new StreamSession(StreamOperation.BOOTSTRAP, peer, null, 0, null, PreviewKind.NONE);
StreamCoordinator streamCoordinator = new StreamCoordinator(StreamOperation.BOOTSTRAP, 1, new DefaultConnectionFactory(), false, false, null, PreviewKind.NONE);
StreamResultFuture future = StreamResultFuture.createInitiator(UUID.randomUUID(), StreamOperation.OTHER, Collections.<StreamEventHandler>emptyList(), streamCoordinator);
StreamSession session = new StreamSession(StreamOperation.BOOTSTRAP, peer, null, false, 0, null, PreviewKind.NONE);
session.init(future);
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD);

6
test/unit/org/apache/cassandra/streaming/async/NettyStreamingMessageSenderTest.java
View File

@ -63,9 +63,11 @@ public class NettyStreamingMessageSenderTest
channel = new TestChannel(Integer.MAX_VALUE);
channel.attr(NettyStreamingMessageSender.TRANSFERRING_FILE_ATTR).set(Boolean.FALSE);
UUID pendingRepair = UUID.randomUUID();
session = new StreamSession(StreamOperation.BOOTSTRAP, REMOTE_ADDR, (template, messagingVersion) -> null, 0, pendingRepair, PreviewKind.ALL);
StreamResultFuture future = StreamResultFuture.initReceivingSide(0, UUID.randomUUID(), StreamOperation.REPAIR, REMOTE_ADDR, channel, pendingRepair, session.getPreviewKind());
session = new StreamSession(StreamOperation.BOOTSTRAP, REMOTE_ADDR, (template, messagingVersion) -> null, true, 0, pendingRepair, PreviewKind.ALL);
StreamResultFuture future = StreamResultFuture.createFollower(0, UUID.randomUUID(), StreamOperation.REPAIR, REMOTE_ADDR, channel, pendingRepair, session.getPreviewKind());
session.init(future);
session.attachOutbound(channel);
sender = session.getMessageSender();
sender.setControlMessageChannel(channel);
}

6
test/unit/org/apache/cassandra/streaming/async/StreamingInboundHandlerTest.java
View File

@ -127,7 +127,7 @@ public class StreamingInboundHandlerTest
private StreamSession createSession(SessionIdentifier sid)
{
return new StreamSession(StreamOperation.BOOTSTRAP, sid.from, (template, messagingVersion) -> null, sid.sessionIndex, UUID.randomUUID(), PreviewKind.ALL);
return new StreamSession(StreamOperation.BOOTSTRAP, sid.from, (template, messagingVersion) -> null, true, sid.sessionIndex, UUID.randomUUID(), PreviewKind.ALL);
}
@Test (expected = IllegalStateException.class)
@ -152,8 +152,8 @@ public class StreamingInboundHandlerTest
public void StreamDeserializingTask_deriveSession_IFM_HasSession()
{
UUID planId = UUID.randomUUID();
StreamResultFuture future = StreamResultFuture.initReceivingSide(0, planId, StreamOperation.REPAIR, REMOTE_ADDR, channel, UUID.randomUUID(), PreviewKind.ALL);
StreamManager.instance.register(future);
StreamResultFuture future = StreamResultFuture.createFollower(0, planId, StreamOperation.REPAIR, REMOTE_ADDR, channel, UUID.randomUUID(), PreviewKind.ALL);
StreamManager.instance.registerFollower(future);
StreamMessageHeader header = new StreamMessageHeader(TableId.generate(), REMOTE_ADDR, planId, 0,
0, 0, UUID.randomUUID());
IncomingStreamMessage msg = new IncomingStreamMessage(null, header);