This is an automated email from the ASF dual-hosted git repository. blerer pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit c17af2d6babe803418b69b9ac7956f4f4bf09c84 Merge: 43c1987 3cda9d7 Author: Benjamin Lerer <b.le...@gmail.com> AuthorDate: Mon May 25 16:34:57 2020 +0200 Merge branch cassandra-3.11 into trunk CHANGES.txt | 1 + .../org/apache/cassandra/streaming/StreamCoordinator.java | 2 ++ .../org/apache/cassandra/streaming/StreamResultFuture.java | 12 +++++++++++- 3 files changed, 14 insertions(+), 1 deletion(-) diff --cc CHANGES.txt index c6c2358,11515c4..01f74f0 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,37 -1,8 +1,38 @@@ -3.11.7 +4.0-alpha5 + * Provide ability to configure IAuditLogger (CASSANDRA-15748) + * Fix nodetool enablefullquerylog blocking param parsing (CASSANDRA-15819) + * Add isTransient to SSTableMetadataView (CASSANDRA-15806) + * Fix tools/bin/fqltool for all shells (CASSANDRA-15820) + * Fix clearing of legacy size_estimates (CASSANDRA-15776) + * Update port when reconnecting to pre-4.0 SSL storage (CASSANDRA-15727) + * Only calculate dynamicBadnessThreshold once per loop in DynamicEndpointSnitch (CASSANDRA-15798) + * Cleanup redundant nodetool commands added in 4.0 (CASSANDRA-15256) + * Update to Python driver 3.23 for cqlsh (CASSANDRA-15793) + * Add tunable initial size and growth factor to RangeTombstoneList (CASSANDRA-15763) + * Improve debug logging in SSTableReader for index summary (CASSANDRA-15755) + * bin/sstableverify should support user provided token ranges (CASSANDRA-15753) + * Improve logging when mutation passed to commit log is too large (CASSANDRA-14781) + * replace LZ4FastDecompressor with LZ4SafeDecompressor (CASSANDRA-15560) + * Fix buffer pool NPE with concurrent release due to in-progress tiny pool eviction (CASSANDRA-15726) + * 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) + * Fix CQLSH UTF-8 encoding issue for Python 2/3 compatibility (CASSANDRA-15739) + * Fix batch statement preparation when multiple tables and parameters are used (CASSANDRA-15730) + * Fix regression with traceOutgoingMessage printing message size (CASSANDRA-15687) + * Ensure repaired data tracking reads a consistent amount of data across replicas (CASSANDRA-15601) + * Fix CQLSH to avoid arguments being evaluated (CASSANDRA-15660) + * Correct Visibility and Improve Safety of Methods in LatencyMetrics (CASSANDRA-15597) + * Allow cqlsh to run with Python2.7/Python3.6+ (CASSANDRA-15659,CASSANDRA-15573) + * Improve logging around incremental repair (CASSANDRA-15599) + * Do not check cdc_raw_directory filesystem space if CDC disabled (CASSANDRA-15688) + * Replace array iterators with get by index (CASSANDRA-15394) + * Minimize BTree iterator allocations (CASSANDRA-15389) +Merged from 3.11: * Fix CQL formatting of read command restrictions for slow query log (CASSANDRA-15503) - * Allow sstableloader to use SSL on the native port (CASSANDRA-14904) Merged from 3.0: + * Rely on snapshotted session infos on StreamResultFuture.maybeComplete to avoid race conditions (CASSANDRA-15667) * EmptyType doesn't override writeValue so could attempt to write bytes when expected not to (CASSANDRA-15790) * Fix index queries on partition key columns when some partitions contains only static data (CASSANDRA-13666) * Avoid creating duplicate rows during major upgrades (CASSANDRA-15789) diff --cc src/java/org/apache/cassandra/streaming/StreamCoordinator.java index c8ebabb,b801ecc..e590e96 --- a/src/java/org/apache/cassandra/streaming/StreamCoordinator.java +++ b/src/java/org/apache/cassandra/streaming/StreamCoordinator.java @@@ -285,9 -294,9 +285,10 @@@ public class StreamCoordinato // create if (streamSessions.size() < connectionsPerHost) { - StreamSession session = new StreamSession(peer, connecting, factory, streamSessions.size(), keepSSTableLevel, isIncremental); + StreamSession session = new StreamSession(streamOperation, peer, factory, isFollower(), streamSessions.size(), + pendingRepair, previewKind); streamSessions.put(++lastReturned, session); + sessionInfos.put(lastReturned, session.getSessionInfo()); return session; } // get @@@ -318,8 -327,9 +319,9 @@@ StreamSession session = streamSessions.get(id); if (session == null) { - session = new StreamSession(peer, connecting, factory, id, keepSSTableLevel, isIncremental); + session = new StreamSession(streamOperation, peer, factory, isFollower(), id, pendingRepair, previewKind); streamSessions.put(id, session); + sessionInfos.put(id, session.getSessionInfo()); } return session; } diff --cc src/java/org/apache/cassandra/streaming/StreamResultFuture.java index 2b5791f,481e93d..76a33a0 --- a/src/java/org/apache/cassandra/streaming/StreamResultFuture.java +++ b/src/java/org/apache/cassandra/streaming/StreamResultFuture.java @@@ -227,8 -222,13 +227,18 @@@ public final class StreamResultFuture e } } + StreamSession getSession(InetAddressAndPort peer, int sessionIndex) + { + return coordinator.getSessionById(peer, sessionIndex); + } ++ + /** + * We can't use {@link StreamCoordinator#hasActiveSessions()} directly because {@link this#maybeComplete()} + * relies on the snapshotted state from {@link StreamCoordinator} and not the {@link StreamSession} state + * directly (CASSANDRA-15667), otherwise inconsistent snapshotted states may lead to completion races. + */ + private boolean finishedAllSessions() + { + return coordinator.getAllSessionInfo().stream().allMatch(s -> s.state.isFinalState()); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org