You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by aw...@apache.org on 2018/01/25 20:12:16 UTC
[19/19] cassandra git commit: Allow storage port to be configurable
per node
Allow storage port to be configurable per node
Patch by Ariel Weisberg; Reviewed by Jason Brown for CASSANDRA-7544
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/59b5b6be
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/59b5b6be
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/59b5b6be
Branch: refs/heads/trunk
Commit: 59b5b6bef0fa76bf5740b688fcd4d9cf525760d0
Parents: 4de7a65
Author: Ariel Weisberg <aw...@apple.com>
Authored: Thu Nov 9 11:33:48 2017 -0500
Committer: Ariel Weisberg <aw...@apple.com>
Committed: Thu Jan 25 14:32:24 2018 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
NEWS.txt | 18 +-
bin/cqlsh.py | 12 +-
build.xml | 10 +-
conf/cassandra.yaml | 9 +-
...iver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar | Bin 2618371 -> 0 bytes
...sandra-driver-core-4.0.0-SNAPSHOT-shaded.jar | Bin 0 -> 2621460 bytes
lib/cassandra-driver-internal-only-3.11.zip | Bin 264882 -> 0 bytes
...iver-internal-only-3.12.0.post0-9ee88ded.zip | Bin 0 -> 265110 bytes
.../cassandra/batchlog/BatchlogManager.java | 52 +-
.../cassandra/config/DatabaseDescriptor.java | 27 +-
.../apache/cassandra/db/ConsistencyLevel.java | 40 +-
.../db/CounterMutationVerbHandler.java | 2 +-
.../cassandra/db/DiskBoundaryManager.java | 4 +-
src/java/org/apache/cassandra/db/Keyspace.java | 1 +
src/java/org/apache/cassandra/db/Mutation.java | 3 -
.../cassandra/db/MutationVerbHandler.java | 43 +-
.../cassandra/db/SizeEstimatesRecorder.java | 4 +-
.../org/apache/cassandra/db/SystemKeyspace.java | 522 +++++++-----
.../cassandra/db/SystemKeyspaceMigrator40.java | 184 +++++
.../org/apache/cassandra/db/view/ViewUtils.java | 18 +-
.../org/apache/cassandra/dht/BootStrapper.java | 12 +-
.../cassandra/dht/RangeFetchMapCalculator.java | 37 +-
.../org/apache/cassandra/dht/RangeStreamer.java | 96 +--
.../dht/tokenallocator/TokenAllocation.java | 50 +-
.../tokenallocator/TokenAllocatorFactory.java | 8 +-
.../exceptions/ReadFailureException.java | 4 +-
.../exceptions/RequestFailureException.java | 6 +-
.../exceptions/WriteFailureException.java | 4 +-
.../apache/cassandra/gms/ApplicationState.java | 11 +-
.../org/apache/cassandra/gms/EndpointState.java | 8 +-
.../apache/cassandra/gms/FailureDetector.java | 74 +-
.../cassandra/gms/FailureDetectorMBean.java | 9 +-
.../org/apache/cassandra/gms/GossipDigest.java | 14 +-
.../apache/cassandra/gms/GossipDigestAck.java | 22 +-
.../apache/cassandra/gms/GossipDigestAck2.java | 22 +-
.../gms/GossipDigestAck2VerbHandler.java | 6 +-
.../gms/GossipDigestAckVerbHandler.java | 10 +-
.../gms/GossipDigestSynVerbHandler.java | 12 +-
src/java/org/apache/cassandra/gms/Gossiper.java | 274 ++++---
.../gms/IEndpointStateChangeSubscriber.java | 16 +-
.../gms/IFailureDetectionEventListener.java | 4 +-
.../apache/cassandra/gms/IFailureDetector.java | 12 +-
.../apache/cassandra/gms/VersionedValue.java | 17 +
.../apache/cassandra/hadoop/ConfigHelper.java | 12 +
.../hadoop/cql3/CqlBulkRecordWriter.java | 20 +-
.../cassandra/hadoop/cql3/CqlConfigHelper.java | 12 +
.../apache/cassandra/hints/HintVerbHandler.java | 6 +-
.../cassandra/hints/HintsDispatchExecutor.java | 12 +-
.../cassandra/hints/HintsDispatchTrigger.java | 2 +-
.../apache/cassandra/hints/HintsDispatcher.java | 10 +-
.../apache/cassandra/hints/HintsService.java | 8 +-
.../org/apache/cassandra/hints/HintsStore.java | 6 +-
.../io/DummyByteVersionedSerializer.java | 55 ++
.../cassandra/io/ShortVersionedSerializer.java | 47 ++
.../cassandra/io/sstable/SSTableLoader.java | 31 +-
.../locator/AbstractEndpointSnitch.java | 19 +-
.../locator/AbstractNetworkTopologySnitch.java | 8 +-
.../locator/AbstractReplicationStrategy.java | 41 +-
.../cassandra/locator/CloudstackSnitch.java | 11 +-
.../locator/DynamicEndpointSnitch.java | 64 +-
.../locator/DynamicEndpointSnitchMBean.java | 2 +
.../cassandra/locator/Ec2MultiRegionSnitch.java | 11 +
.../org/apache/cassandra/locator/Ec2Snitch.java | 11 +-
.../cassandra/locator/EndpointSnitchInfo.java | 9 +-
.../cassandra/locator/GoogleCloudSnitch.java | 11 +-
.../locator/GossipingPropertyFileSnitch.java | 15 +-
.../cassandra/locator/IEndpointSnitch.java | 13 +-
.../cassandra/locator/ILatencySubscriber.java | 4 +-
.../cassandra/locator/InetAddressAndPort.java | 203 +++++
.../apache/cassandra/locator/LocalStrategy.java | 11 +-
.../locator/NetworkTopologyStrategy.java | 23 +-
.../locator/OldNetworkTopologyStrategy.java | 5 +-
.../cassandra/locator/PendingRangeMaps.java | 57 +-
.../cassandra/locator/PropertyFileSnitch.java | 29 +-
.../cassandra/locator/RackInferringSnitch.java | 10 +-
.../locator/ReconnectableSnitchHelper.java | 60 +-
.../apache/cassandra/locator/SeedProvider.java | 3 +-
.../cassandra/locator/SimpleSeedProvider.java | 8 +-
.../apache/cassandra/locator/SimpleSnitch.java | 9 +-
.../cassandra/locator/SimpleStrategy.java | 7 +-
.../apache/cassandra/locator/TokenMetadata.java | 221 +++--
.../cassandra/metrics/ConnectionMetrics.java | 8 +-
.../cassandra/metrics/HintedHandoffMetrics.java | 25 +-
.../cassandra/metrics/HintsServiceMetrics.java | 11 +-
.../cassandra/metrics/MessagingMetrics.java | 4 +-
.../cassandra/metrics/StreamingMetrics.java | 11 +-
.../apache/cassandra/net/BackPressureState.java | 4 +-
.../cassandra/net/BackPressureStrategy.java | 5 +-
.../org/apache/cassandra/net/CallbackInfo.java | 7 +-
.../net/CompactEndpointSerializationHelper.java | 108 ++-
.../cassandra/net/ForwardToContainer.java | 43 +
.../cassandra/net/ForwardToSerializer.java | 86 ++
.../apache/cassandra/net/IAsyncCallback.java | 7 +-
.../net/IAsyncCallbackWithFailure.java | 5 +-
.../org/apache/cassandra/net/IMessageSink.java | 4 +-
.../cassandra/net/MessageDeliveryTask.java | 14 +-
.../org/apache/cassandra/net/MessageIn.java | 74 +-
.../org/apache/cassandra/net/MessageOut.java | 81 +-
.../apache/cassandra/net/MessagingService.java | 271 ++++--
.../cassandra/net/MessagingServiceMBean.java | 26 +
.../org/apache/cassandra/net/ParameterType.java | 69 ++
.../cassandra/net/RateBasedBackPressure.java | 4 +-
.../net/RateBasedBackPressureState.java | 8 +-
.../apache/cassandra/net/WriteCallbackInfo.java | 7 +-
.../cassandra/net/async/HandshakeProtocol.java | 18 +-
.../net/async/InboundHandshakeHandler.java | 7 +-
.../cassandra/net/async/MessageInHandler.java | 21 +-
.../cassandra/net/async/MessageOutHandler.java | 12 +-
.../cassandra/net/async/NettyFactory.java | 15 +-
.../net/async/OutboundConnectionIdentifier.java | 48 +-
.../net/async/OutboundMessagingConnection.java | 13 +-
.../net/async/OutboundMessagingPool.java | 15 +-
.../repair/AsymmetricLocalSyncTask.java | 8 +-
.../repair/AsymmetricRemoteSyncTask.java | 6 +-
.../cassandra/repair/AsymmetricSyncTask.java | 8 +-
.../apache/cassandra/repair/LocalSyncTask.java | 13 +-
.../org/apache/cassandra/repair/NodePair.java | 25 +-
.../apache/cassandra/repair/RemoteSyncTask.java | 4 +-
.../org/apache/cassandra/repair/RepairJob.java | 56 +-
.../repair/RepairMessageVerbHandler.java | 7 +-
.../apache/cassandra/repair/RepairRunnable.java | 31 +-
.../apache/cassandra/repair/RepairSession.java | 38 +-
.../apache/cassandra/repair/SnapshotTask.java | 10 +-
.../cassandra/repair/StreamingRepairTask.java | 17 +-
.../repair/SystemDistributedKeyspace.java | 98 ++-
.../apache/cassandra/repair/TreeResponse.java | 7 +-
.../apache/cassandra/repair/ValidationTask.java | 7 +-
.../org/apache/cassandra/repair/Validator.java | 12 +-
.../repair/asymmetric/DifferenceHolder.java | 16 +-
.../repair/asymmetric/HostDifferences.java | 14 +-
.../asymmetric/IncomingRepairStreamTracker.java | 4 +-
.../repair/asymmetric/PreferedNodeFilter.java | 5 +-
.../repair/asymmetric/ReduceHelper.java | 42 +-
.../repair/asymmetric/StreamFromOptions.java | 16 +-
.../repair/consistent/ConsistentSession.java | 21 +-
.../repair/consistent/CoordinatorSession.java | 24 +-
.../repair/consistent/CoordinatorSessions.java | 4 +-
.../repair/consistent/LocalSessionInfo.java | 6 +-
.../repair/consistent/LocalSessions.java | 82 +-
.../repair/consistent/SyncStatSummary.java | 14 +-
.../repair/messages/AsymmetricSyncRequest.java | 25 +-
.../repair/messages/FinalizePromise.java | 18 +-
.../messages/PrepareConsistentRequest.java | 31 +-
.../messages/PrepareConsistentResponse.java | 17 +-
.../cassandra/repair/messages/SyncComplete.java | 4 +-
.../cassandra/repair/messages/SyncRequest.java | 26 +-
.../cassandra/schema/MigrationManager.java | 24 +-
.../apache/cassandra/schema/MigrationTask.java | 6 +-
.../cassandra/service/AbstractReadExecutor.java | 42 +-
.../service/AbstractWriteResponseHandler.java | 16 +-
.../cassandra/service/ActiveRepairService.java | 64 +-
.../service/BatchlogResponseHandler.java | 6 +-
.../cassandra/service/CassandraDaemon.java | 7 +-
.../apache/cassandra/service/ClientState.java | 2 +-
.../apache/cassandra/service/DataResolver.java | 18 +-
.../DatacenterSyncWriteResponseHandler.java | 8 +-
.../service/DatacenterWriteResponseHandler.java | 8 +-
.../service/IEndpointLifecycleSubscriber.java | 12 +-
.../cassandra/service/LoadBroadcaster.java | 20 +-
.../apache/cassandra/service/ReadCallback.java | 20 +-
.../apache/cassandra/service/StartupChecks.java | 4 +-
.../apache/cassandra/service/StorageProxy.java | 320 ++++----
.../cassandra/service/StorageProxyMBean.java | 3 +-
.../cassandra/service/StorageService.java | 818 ++++++++++++-------
.../cassandra/service/StorageServiceMBean.java | 57 +-
.../apache/cassandra/service/TokenRange.java | 50 +-
.../cassandra/service/WriteResponseHandler.java | 12 +-
.../service/paxos/PrepareCallback.java | 10 +-
.../cassandra/streaming/ProgressInfo.java | 14 +-
.../apache/cassandra/streaming/SessionInfo.java | 12 +-
.../cassandra/streaming/SessionSummary.java | 23 +-
.../cassandra/streaming/StreamCoordinator.java | 24 +-
.../apache/cassandra/streaming/StreamEvent.java | 4 +-
.../cassandra/streaming/StreamManager.java | 10 +-
.../apache/cassandra/streaming/StreamPlan.java | 16 +-
.../cassandra/streaming/StreamResultFuture.java | 15 +-
.../cassandra/streaming/StreamSession.java | 43 +-
.../async/NettyStreamingMessageSender.java | 4 +-
.../async/StreamingInboundHandler.java | 10 +-
.../management/ProgressInfoCompositeData.java | 30 +-
.../SessionCompleteEventCompositeData.java | 8 +-
.../management/SessionInfoCompositeData.java | 44 +-
.../streaming/messages/FileMessageHeader.java | 14 +-
.../streaming/messages/OutgoingFileMessage.java | 2 +-
.../streaming/messages/StreamInitMessage.java | 12 +-
.../tools/BulkLoadConnectionFactory.java | 10 +-
.../org/apache/cassandra/tools/BulkLoader.java | 27 +-
.../apache/cassandra/tools/LoaderOptions.java | 77 +-
.../org/apache/cassandra/tools/NodeProbe.java | 63 +-
.../org/apache/cassandra/tools/NodeTool.java | 26 +
.../tools/nodetool/DescribeCluster.java | 4 +-
.../cassandra/tools/nodetool/DescribeRing.java | 4 +-
.../tools/nodetool/FailureDetectorInfo.java | 2 +-
.../cassandra/tools/nodetool/GetEndpoints.java | 18 +-
.../cassandra/tools/nodetool/GossipInfo.java | 4 +-
.../tools/nodetool/HostStatWithPort.java | 44 +
.../cassandra/tools/nodetool/NetStats.java | 10 +-
.../cassandra/tools/nodetool/RemoveNode.java | 6 +-
.../cassandra/tools/nodetool/RepairAdmin.java | 6 +-
.../apache/cassandra/tools/nodetool/Ring.java | 211 +++--
.../tools/nodetool/SetHostStatWithPort.java | 56 ++
.../apache/cassandra/tools/nodetool/Status.java | 200 +++--
.../cassandra/tracing/ExpiredTraceState.java | 2 +-
.../apache/cassandra/tracing/TraceKeyspace.java | 48 +-
.../apache/cassandra/tracing/TraceState.java | 6 +-
.../cassandra/tracing/TraceStateImpl.java | 4 +-
.../org/apache/cassandra/tracing/Tracing.java | 52 +-
.../apache/cassandra/tracing/TracingImpl.java | 3 +-
.../org/apache/cassandra/transport/Event.java | 21 +-
.../cassandra/transport/ProtocolVersion.java | 9 +-
.../org/apache/cassandra/transport/Server.java | 60 +-
.../transport/messages/ErrorMessage.java | 13 +-
.../org/apache/cassandra/utils/FBUtilities.java | 53 +-
.../apache/cassandra/utils/JMXServerUtils.java | 2 +-
.../org/apache/cassandra/utils/Mx4jTool.java | 2 +-
.../utils/NativeSSTableLoaderClient.java | 46 +-
.../org/apache/cassandra/utils/UUIDGen.java | 2 +-
test/conf/cassandra.yaml | 3 +-
.../serialization/4.0/gms.EndpointState.bin | Bin 73 -> 73 bytes
test/data/serialization/4.0/gms.Gossip.bin | Bin 158 -> 166 bytes
.../serialization/4.0/service.SyncComplete.bin | Bin 538 -> 554 bytes
.../serialization/4.0/service.SyncRequest.bin | Bin 227 -> 241 bytes
.../4.0/service.ValidationComplete.bin | Bin 1251 -> 1257 bytes
.../4.0/service.ValidationRequest.bin | Bin 167 -> 169 bytes
.../locator/DynamicEndpointSnitchLongTest.java | 15 +-
.../cassandra/streaming/LongStreamingTest.java | 4 +-
.../test/microbench/PendingRangesBench.java | 10 +-
.../OffsetAwareConfigurationLoader.java | 32 +
test/unit/org/apache/cassandra/Util.java | 9 +-
.../batchlog/BatchlogEndpointFilterTest.java | 84 +-
.../cassandra/batchlog/BatchlogManagerTest.java | 6 +-
.../config/DatabaseDescriptorRefTest.java | 2 +
.../org/apache/cassandra/cql3/CQLTester.java | 12 +-
.../cassandra/cql3/PreparedStatementsTest.java | 1 -
.../apache/cassandra/cql3/ViewComplexTest.java | 8 +-
.../cql3/validation/operations/CreateTest.java | 8 +-
.../org/apache/cassandra/db/CleanupTest.java | 22 +-
.../cassandra/db/DiskBoundaryManagerTest.java | 6 +-
.../apache/cassandra/db/ReadCommandTest.java | 12 +-
.../org/apache/cassandra/db/RowCacheTest.java | 6 +-
.../db/SystemKeyspaceMigrator40Test.java | 192 +++++
.../apache/cassandra/db/SystemKeyspaceTest.java | 6 +-
.../compaction/AbstractPendingRepairTest.java | 4 +-
.../db/compaction/AntiCompactionTest.java | 4 +-
...tionManagerGetSSTablesForValidationTest.java | 6 +-
.../LeveledCompactionStrategyTest.java | 4 +-
.../apache/cassandra/db/view/ViewUtilsTest.java | 48 +-
.../apache/cassandra/dht/BootStrapperTest.java | 34 +-
.../dht/RangeFetchMapCalculatorTest.java | 116 +--
.../cassandra/dht/StreamStateStoreTest.java | 4 +-
.../apache/cassandra/gms/ArrivalWindowTest.java | 15 +-
.../apache/cassandra/gms/EndpointStateTest.java | 8 +-
.../cassandra/gms/FailureDetectorTest.java | 8 +-
.../apache/cassandra/gms/GossipDigestTest.java | 5 +-
.../org/apache/cassandra/gms/GossiperTest.java | 6 +-
.../gms/PendingRangeCalculatorServiceTest.java | 8 +-
.../cassandra/gms/SerializationsTest.java | 12 +-
.../org/apache/cassandra/hints/HintTest.java | 12 +-
.../cassandra/hints/HintsServiceTest.java | 14 +-
.../io/sstable/CQLSSTableWriterTest.java | 2 +-
.../cassandra/io/sstable/LegacySSTableTest.java | 2 +-
.../cassandra/io/sstable/SSTableLoaderTest.java | 2 +-
.../cassandra/locator/CloudstackSnitchTest.java | 7 +-
.../locator/DynamicEndpointSnitchTest.java | 17 +-
.../apache/cassandra/locator/EC2SnitchTest.java | 8 +-
.../locator/GoogleCloudSnitchTest.java | 7 +-
.../GossipingPropertyFileSnitchTest.java | 15 +-
.../locator/InetAddressAndPortTest.java | 143 ++++
.../locator/NetworkTopologyStrategyTest.java | 73 +-
.../locator/OldNetworkTopologyStrategyTest.java | 21 +-
.../cassandra/locator/PendingRangeMapsTest.java | 37 +-
.../locator/PropertyFileSnitchTest.java | 108 +--
.../locator/ReconnectableSnitchHelperTest.java | 2 +-
.../ReplicationStrategyEndpointCacheTest.java | 35 +-
.../cassandra/locator/SimpleStrategyTest.java | 19 +-
.../cassandra/locator/TokenMetadataTest.java | 49 +-
.../metrics/HintedHandOffMetricsTest.java | 10 +-
.../CompactEndpointSerializationHelperTest.java | 72 ++
.../cassandra/net/ForwardToContainerTest.java | 100 +++
test/unit/org/apache/cassandra/net/Matcher.java | 4 +-
.../apache/cassandra/net/MatcherResponse.java | 11 +-
.../cassandra/net/MessagingServiceTest.java | 82 +-
.../cassandra/net/MockMessagingService.java | 15 +-
.../cassandra/net/MockMessagingServiceTest.java | 6 +-
.../net/RateBasedBackPressureTest.java | 42 +-
.../cassandra/net/WriteCallbackInfoTest.java | 4 +-
.../cassandra/net/async/ChannelWriterTest.java | 6 +-
.../net/async/HandshakeHandlersTest.java | 8 +-
.../net/async/HandshakeProtocolTest.java | 2 +-
.../net/async/InboundHandshakeHandlerTest.java | 18 +-
.../net/async/MessageInHandlerTest.java | 39 +-
.../net/async/MessageOutHandlerTest.java | 16 +-
.../cassandra/net/async/NettyFactoryTest.java | 15 +-
.../net/async/OutboundHandshakeHandlerTest.java | 6 +-
.../async/OutboundMessagingConnectionTest.java | 48 +-
.../net/async/OutboundMessagingPoolTest.java | 11 +-
.../cassandra/repair/AbstractRepairTest.java | 20 +-
.../cassandra/repair/LocalSyncTaskTest.java | 12 +-
.../cassandra/repair/RepairRunnableTest.java | 3 +-
.../cassandra/repair/RepairSessionTest.java | 7 +-
.../apache/cassandra/repair/ValidatorTest.java | 12 +-
.../repair/asymmetric/DifferenceHolderTest.java | 10 +-
.../repair/asymmetric/ReduceHelperTest.java | 74 +-
.../asymmetric/StreamFromOptionsTest.java | 36 +-
.../AbstractConsistentSessionTest.java | 20 +-
.../consistent/CoordinatorSessionTest.java | 30 +-
.../consistent/CoordinatorSessionsTest.java | 12 +-
.../repair/consistent/LocalSessionAccessor.java | 4 +-
.../repair/consistent/LocalSessionTest.java | 17 +-
.../consistent/PendingAntiCompactionTest.java | 6 +-
.../RepairMessageSerializationsTest.java | 12 +-
.../messages/RepairMessageSerializerTest.java | 14 +-
.../service/ActiveRepairServiceTest.java | 58 +-
.../cassandra/service/DataResolverTest.java | 84 +-
.../service/LeaveAndBootstrapTest.java | 63 +-
.../org/apache/cassandra/service/MoveTest.java | 146 ++--
.../service/ProtocolBetaVersionTest.java | 2 +-
.../cassandra/service/ReadExecutorTest.java | 6 +-
.../apache/cassandra/service/RemoveTest.java | 14 +-
.../cassandra/service/SerializationsTest.java | 30 +-
.../cassandra/service/StorageProxyTest.java | 6 +-
.../service/StorageServiceServerTest.java | 170 ++--
.../service/WriteResponseHandlerTest.java | 25 +-
.../cassandra/streaming/SessionInfoTest.java | 4 +-
.../streaming/StreamTransferTaskTest.java | 6 +-
.../streaming/StreamingTransferTest.java | 4 +-
.../async/NettyStreamingMessageSenderTest.java | 8 +-
.../async/StreamingInboundHandlerTest.java | 12 +-
.../apache/cassandra/tracing/TracingTest.java | 3 +-
.../cassandra/transport/ErrorMessageTest.java | 20 +-
.../transport/ProtocolVersionTest.java | 9 +-
.../cassandra/transport/SerDeserTest.java | 12 +-
.../apache/cassandra/utils/FBUtilitiesTest.java | 8 +-
.../cassandra/stress/CompactionStress.java | 4 +-
.../cassandra/stress/settings/SettingsNode.java | 16 +-
.../cassandra/stress/util/JavaDriverClient.java | 7 +-
337 files changed, 6456 insertions(+), 3772 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index b29ae78..d8ae88f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.0
+ * Allow storage port to be configurable per node (CASSANDRA-7544)
* Make sub-range selection for non-frozen collections return null instead of empty (CASSANDRA-14182)
* BloomFilter serialization format should not change byte ordering (CASSANDRA-9067)
* Remove unused on-heap BloomFilter implementation (CASSANDRA-14152)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/NEWS.txt
----------------------------------------------------------------------
diff --git a/NEWS.txt b/NEWS.txt
index 06d73ea..c314030 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -39,7 +39,7 @@ New features
This property can be modified at runtime through both JMX and the new `setconcurrentviewbuilders`
and `getconcurrentviewbuilders` nodetool commands. See CASSANDRA-12245 for more details.
- There is now a binary full query log based on Chronicle Queue that can be controlled using
- nodetool enablefullquerylog, disablefullquerylog, and resetfullquerylog. The log
+ nodetool enablefullquerylog, disablefullquerylog, and resetfullquerylog. The log
contains all queries invoked, approximate time they were invoked, any parameters necessary
to bind wildcard values, and all query options. A human readable version of the log can be
dumped or tailed using the new bin/fqltool utility. The full query log is designed to be safe
@@ -103,6 +103,22 @@ Upgrading
but blocks for up to a configurable number of milliseconds between disk flushes.
- nodetool clearsnapshot now required the --all flag to remove all snapshots.
Previous behavior would delete all snapshots by default.
+ - Nodes are now identified by a combination of IP, and storage port.
+ Existing JMX APIs, nodetool, and system tables continue to work
+ and accept/return just an IP, but there is a new
+ version of each that works with the full unambiguous identifier.
+ You should prefer these over the deprecated ambiguous versions that only
+ work with an IP. This was done to support multiple instances per IP.
+ Additionally we are moving to only using a single port for encrypted and
+ unencrypted traffic and if you want multiple instances per IP you must
+ first switch encrypted traffic to the storage port and not a separate
+ encrypted port. If you want to use multiple instances per IP
+ with SSL you will need to use StartTLS on storage_port and set
+ outgoing_encrypted_port_source to gossip outbound connections
+ know what port to connect to for each instance. Before changing
+ storage port or native port at nodes you must first upgrade the entire cluster
+ and clients to 4.0 so they can handle the port not being consistent across
+ the cluster.
Materialized Views
-------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/bin/cqlsh.py
----------------------------------------------------------------------
diff --git a/bin/cqlsh.py b/bin/cqlsh.py
index 61ea160..3055110 100644
--- a/bin/cqlsh.py
+++ b/bin/cqlsh.py
@@ -455,7 +455,8 @@ class Shell(cmd.Cmd):
single_statement=None,
request_timeout=DEFAULT_REQUEST_TIMEOUT_SECONDS,
protocol_version=None,
- connect_timeout=DEFAULT_CONNECT_TIMEOUT_SECONDS):
+ connect_timeout=DEFAULT_CONNECT_TIMEOUT_SECONDS,
+ allow_server_port_discovery=False):
cmd.Cmd.__init__(self, completekey=completekey)
self.hostname = hostname
self.port = port
@@ -470,6 +471,7 @@ class Shell(cmd.Cmd):
self.tracing_enabled = tracing_enabled
self.page_size = self.default_page_size
self.expand_enabled = expand_enabled
+ self.allow_server_port_discovery = allow_server_port_discovery
if use_conn:
self.conn = use_conn
else:
@@ -482,6 +484,7 @@ class Shell(cmd.Cmd):
load_balancing_policy=WhiteListRoundRobinPolicy([self.hostname]),
control_connection_timeout=connect_timeout,
connect_timeout=connect_timeout,
+ allow_server_port_discovery=allow_server_port_discovery,
**kwargs)
self.owns_connection = not use_conn
@@ -1768,7 +1771,8 @@ class Shell(cmd.Cmd):
display_timezone=self.display_timezone,
max_trace_wait=self.max_trace_wait, ssl=self.ssl,
request_timeout=self.session.default_timeout,
- connect_timeout=self.conn.connect_timeout)
+ connect_timeout=self.conn.connect_timeout,
+ allow_server_port_discovery=self.allow_server_port_discovery)
subshell.cmdloop()
f.close()
@@ -2252,6 +2256,7 @@ def read_options(cmdlineargs, environment):
optvalues.connect_timeout = option_with_default(configs.getint, 'connection', 'timeout', DEFAULT_CONNECT_TIMEOUT_SECONDS)
optvalues.request_timeout = option_with_default(configs.getint, 'connection', 'request_timeout', DEFAULT_REQUEST_TIMEOUT_SECONDS)
optvalues.execute = None
+ optvalues.allow_server_port_discovery = option_with_default(configs.getboolean, 'connection', 'allow_server_port_discovery', 'False')
(options, arguments) = parser.parse_args(cmdlineargs, values=optvalues)
@@ -2415,7 +2420,8 @@ def main(options, hostname, port):
single_statement=options.execute,
request_timeout=options.request_timeout,
connect_timeout=options.connect_timeout,
- encoding=options.encoding)
+ encoding=options.encoding,
+ allow_server_port_discovery=options.allow_server_port_discovery)
except KeyboardInterrupt:
sys.exit('Connection aborted.')
except CQL_ERRORS, e:
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index 8a033e0..5796868 100644
--- a/build.xml
+++ b/build.xml
@@ -306,7 +306,7 @@
<!-- define the remote repositories we use -->
<artifact:remoteRepository id="central" url="${artifact.remoteRepository.central}"/>
<artifact:remoteRepository id="apache" url="${artifact.remoteRepository.apache}"/>
-
+
<macrodef name="install">
<attribute name="pomFile"/>
<attribute name="file"/>
@@ -437,7 +437,7 @@
<dependency groupId="com.google.code.findbugs" artifactId="jsr305" version="2.0.2" />
<dependency groupId="com.clearspring.analytics" artifactId="stream" version="2.5.2" />
<!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
- <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="3.0.1" classifier="shaded">
+ <dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" version="4.0.0-SNAPSHOT" classifier="shaded">
<exclusion groupId="io.netty" artifactId="netty-buffer"/>
<exclusion groupId="io.netty" artifactId="netty-codec"/>
<exclusion groupId="io.netty" artifactId="netty-handler"/>
@@ -522,7 +522,7 @@
<dependency groupId="org.apache.hadoop" artifactId="hadoop-minicluster"/>
<dependency groupId="com.google.code.findbugs" artifactId="jsr305"/>
<dependency groupId="org.antlr" artifactId="antlr"/>
- <!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
+ <!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
<dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded"/>
-->
<dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>
@@ -543,7 +543,7 @@
<dependency groupId="junit" artifactId="junit"/>
<!-- UPDATE AND UNCOMMENT ON THE DRIVER RELEASE, BEFORE 4.0 RELEASE
<dependency groupId="com.datastax.cassandra" artifactId="cassandra-driver-core" classifier="shaded"/>
- -->
+ -->
<dependency groupId="io.netty" artifactId="netty-all"/>
<dependency groupId="org.eclipse.jdt.core.compiler" artifactId="ecj"/>
<dependency groupId="org.caffinitas.ohc" artifactId="ohc-core"/>
@@ -624,7 +624,7 @@
<exclusion groupId="io.netty" artifactId="netty-transport"/>
</dependency>
-->
-
+
<!-- don't need jna to run, but nice to have -->
<dependency groupId="net.java.dev.jna" artifactId="jna"/>
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/conf/cassandra.yaml
----------------------------------------------------------------------
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index 80e4515..3bed3a6 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -416,7 +416,7 @@ seed_provider:
parameters:
# seeds is actually a comma-delimited list of addresses.
# Ex: "<ip1>,<ip2>,<ip3>"
- - seeds: "127.0.0.1"
+ - seeds: "127.0.0.1:7000"
# For workloads with more data than can fit in memory, Cassandra's
# bottleneck will be reads that need to fetch data from
@@ -945,6 +945,13 @@ dynamic_snitch_badness_threshold: 0.1
# the keystore and truststore. For instructions on generating these files, see:
# http://download.oracle.com/javase/8/docs/technotes/guides/security/jsse/JSSERefGuide.html#CreateKeystore
#
+# If you are taking advantage of StartTLS outbound connections will have the issue that they can't know
+# what encrypted port to connect to in a foolproof way. outgoing_encrypted_port_source deals with this confusion
+# by allowing you to specify how you want a node to pick an outgoing port for intra-cluster connections.
+# Valid values are "gossip" and "yaml". Gossip will always connect to the storage port for a node that is
+# published via a gossip which is always going to be the plain storage port. "yaml" will always select
+# the port configured as ssl_storage_port on THIS node. If you want to use SSL and have different storage
+# ports across the cluster you must select "gossip" and use StartTLS on storage_port.
server_encryption_options:
# set to true for allowing secure incoming connections
enabled: false
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar
----------------------------------------------------------------------
diff --git a/lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar b/lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar
deleted file mode 100644
index a7be9cb..0000000
Binary files a/lib/cassandra-driver-core-3.3.2-0461ed35-SNAPSHOT-shaded.jar and /dev/null differ
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar
----------------------------------------------------------------------
diff --git a/lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar b/lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar
new file mode 100644
index 0000000..609c393
Binary files /dev/null and b/lib/cassandra-driver-core-4.0.0-SNAPSHOT-shaded.jar differ
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/lib/cassandra-driver-internal-only-3.11.zip
----------------------------------------------------------------------
diff --git a/lib/cassandra-driver-internal-only-3.11.zip b/lib/cassandra-driver-internal-only-3.11.zip
deleted file mode 100644
index f7760af..0000000
Binary files a/lib/cassandra-driver-internal-only-3.11.zip and /dev/null differ
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip
----------------------------------------------------------------------
diff --git a/lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip b/lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip
new file mode 100644
index 0000000..4aa91b7
Binary files /dev/null and b/lib/cassandra-driver-internal-only-3.12.0.post0-9ee88ded.zip differ
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
index 807e970..f232bdc 100644
--- a/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
+++ b/src/java/org/apache/cassandra/batchlog/BatchlogManager.java
@@ -19,7 +19,6 @@ package org.apache.cassandra.batchlog;
import java.io.IOException;
import java.lang.management.ManagementFactory;
-import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
@@ -34,6 +33,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.*;
@@ -250,7 +250,7 @@ public class BatchlogManager implements BatchlogManagerMBean
int positionInPage = 0;
ArrayList<ReplayingBatch> unfinishedBatches = new ArrayList<>(pageSize);
- Set<InetAddress> hintedNodes = new HashSet<>();
+ Set<InetAddressAndPort> hintedNodes = new HashSet<>();
Set<UUID> replayedBatches = new HashSet<>();
// Sending out batches for replay without waiting for them, so that one stuck batch doesn't affect others
@@ -295,7 +295,7 @@ public class BatchlogManager implements BatchlogManagerMBean
replayedBatches.forEach(BatchlogManager::remove);
}
- private void finishAndClearBatches(ArrayList<ReplayingBatch> batches, Set<InetAddress> hintedNodes, Set<UUID> replayedBatches)
+ private void finishAndClearBatches(ArrayList<ReplayingBatch> batches, Set<InetAddressAndPort> hintedNodes, Set<UUID> replayedBatches)
{
// schedule hints for timed out deliveries
for (ReplayingBatch batch : batches)
@@ -330,7 +330,7 @@ public class BatchlogManager implements BatchlogManagerMBean
this.replayedBytes = addMutations(version, serializedMutations);
}
- public int replay(RateLimiter rateLimiter, Set<InetAddress> hintedNodes) throws IOException
+ public int replay(RateLimiter rateLimiter, Set<InetAddressAndPort> hintedNodes) throws IOException
{
logger.trace("Replaying batch {}", id);
@@ -348,7 +348,7 @@ public class BatchlogManager implements BatchlogManagerMBean
return replayHandlers.size();
}
- public void finish(Set<InetAddress> hintedNodes)
+ public void finish(Set<InetAddressAndPort> hintedNodes)
{
for (int i = 0; i < replayHandlers.size(); i++)
{
@@ -396,7 +396,7 @@ public class BatchlogManager implements BatchlogManagerMBean
mutations.add(mutation);
}
- private void writeHintsForUndeliveredEndpoints(int startFrom, Set<InetAddress> hintedNodes)
+ private void writeHintsForUndeliveredEndpoints(int startFrom, Set<InetAddressAndPort> hintedNodes)
{
int gcgs = gcgs(mutations);
@@ -420,7 +420,7 @@ public class BatchlogManager implements BatchlogManagerMBean
private static List<ReplayWriteResponseHandler<Mutation>> sendReplays(List<Mutation> mutations,
long writtenAt,
- Set<InetAddress> hintedNodes)
+ Set<InetAddressAndPort> hintedNodes)
{
List<ReplayWriteResponseHandler<Mutation>> handlers = new ArrayList<>(mutations.size());
for (Mutation mutation : mutations)
@@ -440,15 +440,15 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
private static ReplayWriteResponseHandler<Mutation> sendSingleReplayMutation(final Mutation mutation,
long writtenAt,
- Set<InetAddress> hintedNodes)
+ Set<InetAddressAndPort> hintedNodes)
{
- Set<InetAddress> liveEndpoints = new HashSet<>();
+ Set<InetAddressAndPort> liveEndpoints = new HashSet<>();
String ks = mutation.getKeyspaceName();
Token tk = mutation.key().getToken();
- for (InetAddress endpoint : StorageService.instance.getNaturalAndPendingEndpoints(ks, tk))
+ for (InetAddressAndPort endpoint : StorageService.instance.getNaturalAndPendingEndpoints(ks, tk))
{
- if (endpoint.equals(FBUtilities.getBroadcastAddress()))
+ if (endpoint.equals(FBUtilities.getBroadcastAddressAndPort()))
{
mutation.apply();
}
@@ -469,7 +469,7 @@ public class BatchlogManager implements BatchlogManagerMBean
ReplayWriteResponseHandler<Mutation> handler = new ReplayWriteResponseHandler<>(liveEndpoints, System.nanoTime());
MessageOut<Mutation> message = mutation.createMessage();
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
MessagingService.instance().sendRR(message, endpoint, handler, false);
return handler;
}
@@ -488,11 +488,11 @@ public class BatchlogManager implements BatchlogManagerMBean
*/
private static class ReplayWriteResponseHandler<T> extends WriteResponseHandler<T>
{
- private final Set<InetAddress> undelivered = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final Set<InetAddressAndPort> undelivered = Collections.newSetFromMap(new ConcurrentHashMap<>());
- ReplayWriteResponseHandler(Collection<InetAddress> writeEndpoints, long queryStartNanoTime)
+ ReplayWriteResponseHandler(Collection<InetAddressAndPort> writeEndpoints, long queryStartNanoTime)
{
- super(writeEndpoints, Collections.<InetAddress>emptySet(), null, null, null, WriteType.UNLOGGED_BATCH, queryStartNanoTime);
+ super(writeEndpoints, Collections.<InetAddressAndPort>emptySet(), null, null, null, WriteType.UNLOGGED_BATCH, queryStartNanoTime);
undelivered.addAll(writeEndpoints);
}
@@ -505,7 +505,7 @@ public class BatchlogManager implements BatchlogManagerMBean
@Override
public void response(MessageIn<T> m)
{
- boolean removed = undelivered.remove(m == null ? FBUtilities.getBroadcastAddress() : m.from);
+ boolean removed = undelivered.remove(m == null ? FBUtilities.getBroadcastAddressAndPort() : m.from);
assert removed;
super.response(m);
}
@@ -515,9 +515,9 @@ public class BatchlogManager implements BatchlogManagerMBean
public static class EndpointFilter
{
private final String localRack;
- private final Multimap<String, InetAddress> endpoints;
+ private final Multimap<String, InetAddressAndPort> endpoints;
- public EndpointFilter(String localRack, Multimap<String, InetAddress> endpoints)
+ public EndpointFilter(String localRack, Multimap<String, InetAddressAndPort> endpoints)
{
this.localRack = localRack;
this.endpoints = endpoints;
@@ -526,15 +526,15 @@ public class BatchlogManager implements BatchlogManagerMBean
/**
* @return list of candidates for batchlog hosting. If possible these will be two nodes from different racks.
*/
- public Collection<InetAddress> filter()
+ public Collection<InetAddressAndPort> filter()
{
// special case for single-node data centers
if (endpoints.values().size() == 1)
return endpoints.values();
// strip out dead endpoints and localhost
- ListMultimap<String, InetAddress> validated = ArrayListMultimap.create();
- for (Map.Entry<String, InetAddress> entry : endpoints.entries())
+ ListMultimap<String, InetAddressAndPort> validated = ArrayListMultimap.create();
+ for (Map.Entry<String, InetAddressAndPort> entry : endpoints.entries())
if (isValid(entry.getValue()))
validated.put(entry.getKey(), entry.getValue());
@@ -554,7 +554,7 @@ public class BatchlogManager implements BatchlogManagerMBean
* pick two random nodes from there; we are guaranteed to have at least two nodes in the single remaining rack
* because of the preceding if block.
*/
- List<InetAddress> otherRack = Lists.newArrayList(validated.values());
+ List<InetAddressAndPort> otherRack = Lists.newArrayList(validated.values());
shuffle(otherRack);
return otherRack.subList(0, 2);
}
@@ -572,10 +572,10 @@ public class BatchlogManager implements BatchlogManagerMBean
}
// grab a random member of up to two racks
- List<InetAddress> result = new ArrayList<>(2);
+ List<InetAddressAndPort> result = new ArrayList<>(2);
for (String rack : Iterables.limit(racks, 2))
{
- List<InetAddress> rackMembers = validated.get(rack);
+ List<InetAddressAndPort> rackMembers = validated.get(rack);
result.add(rackMembers.get(getRandomInt(rackMembers.size())));
}
@@ -583,9 +583,9 @@ public class BatchlogManager implements BatchlogManagerMBean
}
@VisibleForTesting
- protected boolean isValid(InetAddress input)
+ protected boolean isValid(InetAddressAndPort input)
{
- return !input.equals(FBUtilities.getBroadcastAddress()) && FailureDetector.instance.isAlive(input);
+ return !input.equals(FBUtilities.getBroadcastAddressAndPort()) && FailureDetector.instance.isAlive(input);
}
@VisibleForTesting
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index a656d1f..9012e3a 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -55,6 +55,7 @@ import org.apache.cassandra.io.util.SsdDiskOptimizationStrategy;
import org.apache.cassandra.locator.DynamicEndpointSnitch;
import org.apache.cassandra.locator.EndpointSnitchInfo;
import org.apache.cassandra.locator.IEndpointSnitch;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.SeedProvider;
import org.apache.cassandra.net.BackPressureStrategy;
import org.apache.cassandra.net.RateBasedBackPressure;
@@ -110,7 +111,7 @@ public class DatabaseDescriptor
private static long indexSummaryCapacityInMB;
private static String localDC;
- private static Comparator<InetAddress> localComparator;
+ private static Comparator<InetAddressAndPort> localComparator;
private static EncryptionContext encryptionContext;
private static boolean hasLoggedConfig;
@@ -307,6 +308,7 @@ public class DatabaseDescriptor
private static void applyAll() throws ConfigurationException
{
+ //InetAddressAndPort cares that applySimpleConfig runs first
applySimpleConfig();
applyPartitioner();
@@ -324,6 +326,9 @@ public class DatabaseDescriptor
private static void applySimpleConfig()
{
+ //Doing this first before all other things in case other pieces of config want to construct
+ //InetAddressAndPort and get the right defaults
+ InetAddressAndPort.initializeDefaultPort(getStoragePort());
if (conf.commitlog_sync == null)
{
@@ -827,7 +832,7 @@ public class DatabaseDescriptor
}
else
{
- rpcAddress = FBUtilities.getLocalAddress();
+ rpcAddress = FBUtilities.getJustLocalAddress();
}
/* RPC address to broadcast */
@@ -956,10 +961,10 @@ public class DatabaseDescriptor
snitch = createEndpointSnitch(conf.dynamic_snitch, conf.endpoint_snitch);
EndpointSnitchInfo.create();
- localDC = snitch.getDatacenter(FBUtilities.getBroadcastAddress());
- localComparator = new Comparator<InetAddress>()
+ localDC = snitch.getDatacenter(FBUtilities.getBroadcastAddressAndPort());
+ localComparator = new Comparator<InetAddressAndPort>()
{
- public int compare(InetAddress endpoint1, InetAddress endpoint2)
+ public int compare(InetAddressAndPort endpoint1, InetAddressAndPort endpoint2)
{
boolean local1 = localDC.equals(snitch.getDatacenter(endpoint1));
boolean local2 = localDC.equals(snitch.getDatacenter(endpoint2));
@@ -1319,14 +1324,14 @@ public class DatabaseDescriptor
return conf.num_tokens;
}
- public static InetAddress getReplaceAddress()
+ public static InetAddressAndPort getReplaceAddress()
{
try
{
if (System.getProperty(Config.PROPERTY_PREFIX + "replace_address", null) != null)
- return InetAddress.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address", null));
+ return InetAddressAndPort.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address", null));
else if (System.getProperty(Config.PROPERTY_PREFIX + "replace_address_first_boot", null) != null)
- return InetAddress.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address_first_boot", null));
+ return InetAddressAndPort.getByName(System.getProperty(Config.PROPERTY_PREFIX + "replace_address_first_boot", null));
return null;
}
catch (UnknownHostException e)
@@ -1651,9 +1656,9 @@ public class DatabaseDescriptor
return conf.saved_caches_directory;
}
- public static Set<InetAddress> getSeeds()
+ public static Set<InetAddressAndPort> getSeeds()
{
- return ImmutableSet.<InetAddress>builder().addAll(seedProvider.getSeeds()).build();
+ return ImmutableSet.<InetAddressAndPort>builder().addAll(seedProvider.getSeeds()).build();
}
public static InetAddress getListenAddress()
@@ -2206,7 +2211,7 @@ public class DatabaseDescriptor
return localDC;
}
- public static Comparator<InetAddress> getLocalComparator()
+ public static Comparator<InetAddressAndPort> getLocalComparator()
{
return localComparator;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/ConsistencyLevel.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
index 2214d8d..f93e737 100644
--- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java
+++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
@@ -17,7 +17,6 @@
*/
package org.apache.cassandra.db;
-import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -28,6 +27,7 @@ import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.service.ReadRepairDecision;
@@ -145,21 +145,21 @@ public enum ConsistencyLevel
return isDCLocal;
}
- public boolean isLocal(InetAddress endpoint)
+ public boolean isLocal(InetAddressAndPort endpoint)
{
return DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint));
}
- public int countLocalEndpoints(Iterable<InetAddress> liveEndpoints)
+ public int countLocalEndpoints(Iterable<InetAddressAndPort> liveEndpoints)
{
int count = 0;
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
if (isLocal(endpoint))
count++;
return count;
}
- private Map<String, Integer> countPerDCEndpoints(Keyspace keyspace, Iterable<InetAddress> liveEndpoints)
+ private Map<String, Integer> countPerDCEndpoints(Keyspace keyspace, Iterable<InetAddressAndPort> liveEndpoints)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
@@ -167,7 +167,7 @@ public enum ConsistencyLevel
for (String dc: strategy.getDatacenters())
dcEndpoints.put(dc, 0);
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
{
String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint);
dcEndpoints.put(dc, dcEndpoints.get(dc) + 1);
@@ -175,12 +175,12 @@ public enum ConsistencyLevel
return dcEndpoints;
}
- public List<InetAddress> filterForQuery(Keyspace keyspace, List<InetAddress> liveEndpoints)
+ public List<InetAddressAndPort> filterForQuery(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints)
{
return filterForQuery(keyspace, liveEndpoints, ReadRepairDecision.NONE);
}
- public List<InetAddress> filterForQuery(Keyspace keyspace, List<InetAddress> liveEndpoints, ReadRepairDecision readRepair)
+ public List<InetAddressAndPort> filterForQuery(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints, ReadRepairDecision readRepair)
{
/*
* If we are doing an each quorum query, we have to make sure that the endpoints we select
@@ -206,9 +206,9 @@ public enum ConsistencyLevel
case GLOBAL:
return liveEndpoints;
case DC_LOCAL:
- List<InetAddress> local = new ArrayList<InetAddress>();
- List<InetAddress> other = new ArrayList<InetAddress>();
- for (InetAddress add : liveEndpoints)
+ List<InetAddressAndPort> local = new ArrayList<>();
+ List<InetAddressAndPort> other = new ArrayList<>();
+ for (InetAddressAndPort add : liveEndpoints)
{
if (isLocal(add))
local.add(add);
@@ -225,7 +225,7 @@ public enum ConsistencyLevel
}
}
- private List<InetAddress> filterForEachQuorum(Keyspace keyspace, List<InetAddress> liveEndpoints, ReadRepairDecision readRepair)
+ private List<InetAddressAndPort> filterForEachQuorum(Keyspace keyspace, List<InetAddressAndPort> liveEndpoints, ReadRepairDecision readRepair)
{
NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
@@ -233,20 +233,20 @@ public enum ConsistencyLevel
if (readRepair == ReadRepairDecision.GLOBAL)
return liveEndpoints;
- Map<String, List<InetAddress>> dcsEndpoints = new HashMap<>();
+ Map<String, List<InetAddressAndPort>> dcsEndpoints = new HashMap<>();
for (String dc: strategy.getDatacenters())
dcsEndpoints.put(dc, new ArrayList<>());
- for (InetAddress add : liveEndpoints)
+ for (InetAddressAndPort add : liveEndpoints)
{
String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(add);
dcsEndpoints.get(dc).add(add);
}
- List<InetAddress> waitSet = new ArrayList<>();
- for (Map.Entry<String, List<InetAddress>> dcEndpoints : dcsEndpoints.entrySet())
+ List<InetAddressAndPort> waitSet = new ArrayList<>();
+ for (Map.Entry<String, List<InetAddressAndPort>> dcEndpoints : dcsEndpoints.entrySet())
{
- List<InetAddress> dcEndpoint = dcEndpoints.getValue();
+ List<InetAddressAndPort> dcEndpoint = dcEndpoints.getValue();
if (readRepair == ReadRepairDecision.DC_LOCAL && dcEndpoints.getKey().equals(DatabaseDescriptor.getLocalDataCenter()))
waitSet.addAll(dcEndpoint);
else
@@ -256,7 +256,7 @@ public enum ConsistencyLevel
return waitSet;
}
- public boolean isSufficientLiveNodes(Keyspace keyspace, Iterable<InetAddress> liveEndpoints)
+ public boolean isSufficientLiveNodes(Keyspace keyspace, Iterable<InetAddressAndPort> liveEndpoints)
{
switch (this)
{
@@ -283,7 +283,7 @@ public enum ConsistencyLevel
}
}
- public void assureSufficientLiveNodes(Keyspace keyspace, Iterable<InetAddress> liveEndpoints) throws UnavailableException
+ public void assureSufficientLiveNodes(Keyspace keyspace, Iterable<InetAddressAndPort> liveEndpoints) throws UnavailableException
{
int blockFor = blockFor(keyspace);
switch (this)
@@ -302,7 +302,7 @@ public enum ConsistencyLevel
if (logger.isTraceEnabled())
{
StringBuilder builder = new StringBuilder("Local replicas [");
- for (InetAddress endpoint : liveEndpoints)
+ for (InetAddressAndPort endpoint : liveEndpoints)
{
if (isLocal(endpoint))
builder.append(endpoint).append(",");
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java b/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
index bd273e4..95d7916 100644
--- a/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java
@@ -37,7 +37,7 @@ public class CounterMutationVerbHandler implements IVerbHandler<CounterMutation>
final CounterMutation cm = message.payload;
logger.trace("Applying forwarded {}", cm);
- String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
+ String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddressAndPort());
// We should not wait for the result of the write in this thread,
// otherwise we could have a distributed deadlock between replicas
// running this VerbHandler (see #4578).
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/DiskBoundaryManager.java b/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
index 03cbf7b..72b5e2a 100644
--- a/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
+++ b/src/java/org/apache/cassandra/db/DiskBoundaryManager.java
@@ -80,14 +80,14 @@ public class DiskBoundaryManager
&& !StorageService.isReplacingSameAddress()) // When replacing same address, the node marks itself as UN locally
{
PendingRangeCalculatorService.instance.blockUntilFinished();
- localRanges = tmd.getPendingRanges(cfs.keyspace.getName(), FBUtilities.getBroadcastAddress());
+ localRanges = tmd.getPendingRanges(cfs.keyspace.getName(), FBUtilities.getBroadcastAddressAndPort());
}
else
{
// Reason we use use the future settled TMD is that if we decommission a node, we want to stream
// from that node to the correct location on disk, if we didn't, we would put new files in the wrong places.
// We do this to minimize the amount of data we need to move in rebalancedisks once everything settled
- localRanges = cfs.keyspace.getReplicationStrategy().getAddressRanges(tmd.cloneAfterAllSettled()).get(FBUtilities.getBroadcastAddress());
+ localRanges = cfs.keyspace.getReplicationStrategy().getAddressRanges(tmd.cloneAfterAllSettled()).get(FBUtilities.getBroadcastAddressAndPort());
}
logger.debug("Got local ranges {} (ringVersion = {})", localRanges, ringVersion);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/Keyspace.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java
index cebf6eb..ae778f1 100644
--- a/src/java/org/apache/cassandra/db/Keyspace.java
+++ b/src/java/org/apache/cassandra/db/Keyspace.java
@@ -352,6 +352,7 @@ public class Keyspace
private void createReplicationStrategy(KeyspaceMetadata ksm)
{
+ logger.info("Creating replication strategy " + ksm .name + " params " + ksm.params);
replicationStrategy = AbstractReplicationStrategy.createReplicationStrategy(ksm.name,
ksm.params.replication.klass,
StorageService.instance.getTokenMetadata(),
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/Mutation.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java
index a6a920c..6195fe4 100644
--- a/src/java/org/apache/cassandra/db/Mutation.java
+++ b/src/java/org/apache/cassandra/db/Mutation.java
@@ -43,9 +43,6 @@ public class Mutation implements IMutation
{
public static final MutationSerializer serializer = new MutationSerializer();
- public static final String FORWARD_TO = "FWD_TO";
- public static final String FORWARD_FROM = "FWD_FRM";
-
// todo this is redundant
// when we remove it, also restore SerializationsTest.testMutationRead to not regenerate new Mutations each test
private final String keyspaceName;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/MutationVerbHandler.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/MutationVerbHandler.java b/src/java/org/apache/cassandra/db/MutationVerbHandler.java
index 59247a2..8386048 100644
--- a/src/java/org/apache/cassandra/db/MutationVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/MutationVerbHandler.java
@@ -17,18 +17,17 @@
*/
package org.apache.cassandra.db;
-import java.io.DataInputStream;
import java.io.IOException;
-import java.net.InetAddress;
+import java.util.Iterator;
import org.apache.cassandra.exceptions.WriteTimeoutException;
-import org.apache.cassandra.io.util.FastByteArrayInputStream;
+import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.*;
import org.apache.cassandra.tracing.Tracing;
public class MutationVerbHandler implements IVerbHandler<Mutation>
{
- private void reply(int id, InetAddress replyTo)
+ private void reply(int id, InetAddressAndPort replyTo)
{
Tracing.trace("Enqueuing response to {}", replyTo);
MessagingService.instance().sendReply(WriteResponse.createMessage(), id, replyTo);
@@ -42,18 +41,19 @@ public class MutationVerbHandler implements IVerbHandler<Mutation>
public void doVerb(MessageIn<Mutation> message, int id) throws IOException
{
// Check if there were any forwarding headers in this message
- byte[] from = message.parameters.get(Mutation.FORWARD_FROM);
- InetAddress replyTo;
+ InetAddressAndPort from = (InetAddressAndPort)message.parameters.get(ParameterType.FORWARD_FROM);
+ InetAddressAndPort replyTo;
if (from == null)
{
replyTo = message.from;
- byte[] forwardBytes = message.parameters.get(Mutation.FORWARD_TO);
- if (forwardBytes != null)
- forwardToLocalNodes(message.payload, message.verb, forwardBytes, message.from);
+ ForwardToContainer forwardTo = (ForwardToContainer)message.parameters.get(ParameterType.FORWARD_TO);
+ if (forwardTo != null)
+ forwardToLocalNodes(message.payload, message.verb, forwardTo, message.from);
}
else
{
- replyTo = InetAddress.getByAddress(from);
+
+ replyTo = from;
}
try
@@ -69,22 +69,17 @@ public class MutationVerbHandler implements IVerbHandler<Mutation>
}
}
- private static void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, byte[] forwardBytes, InetAddress from) throws IOException
+ private static void forwardToLocalNodes(Mutation mutation, MessagingService.Verb verb, ForwardToContainer forwardTo, InetAddressAndPort from) throws IOException
{
- try (DataInputStream in = new DataInputStream(new FastByteArrayInputStream(forwardBytes)))
+ // tell the recipients who to send their ack to
+ MessageOut<Mutation> message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(ParameterType.FORWARD_FROM, from);
+ Iterator<InetAddressAndPort> iterator = forwardTo.targets.iterator();
+ // Send a message to each of the addresses on our Forward List
+ for (int i = 0; i < forwardTo.targets.size(); i++)
{
- int size = in.readInt();
-
- // tell the recipients who to send their ack to
- MessageOut<Mutation> message = new MessageOut<>(verb, mutation, Mutation.serializer).withParameter(Mutation.FORWARD_FROM, from.getAddress());
- // Send a message to each of the addresses on our Forward List
- for (int i = 0; i < size; i++)
- {
- InetAddress address = CompactEndpointSerializationHelper.deserialize(in);
- int id = in.readInt();
- Tracing.trace("Enqueuing forwarded write to {}", address);
- MessagingService.instance().sendOneWay(message, id, address);
- }
+ InetAddressAndPort address = iterator.next();
+ Tracing.trace("Enqueuing forwarded write to {}", address);
+ MessagingService.instance().sendOneWay(message, forwardTo.messageIds[i], address);
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/59b5b6be/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
index 066b2fe..151e7d3 100644
--- a/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
+++ b/src/java/org/apache/cassandra/db/SizeEstimatesRecorder.java
@@ -60,7 +60,7 @@ public class SizeEstimatesRecorder extends SchemaChangeListener implements Runna
public void run()
{
TokenMetadata metadata = StorageService.instance.getTokenMetadata().cloneOnlyTokenMap();
- if (!metadata.isMember(FBUtilities.getBroadcastAddress()))
+ if (!metadata.isMember(FBUtilities.getBroadcastAddressAndPort()))
{
logger.debug("Node is not part of the ring; not recording size estimates");
return;
@@ -71,7 +71,7 @@ public class SizeEstimatesRecorder extends SchemaChangeListener implements Runna
for (Keyspace keyspace : Keyspace.nonLocalStrategy())
{
Collection<Range<Token>> localRanges = StorageService.instance.getPrimaryRangesForEndpoint(keyspace.getName(),
- FBUtilities.getBroadcastAddress());
+ FBUtilities.getBroadcastAddressAndPort());
for (ColumnFamilyStore table : keyspace.getColumnFamilyStores())
{
long start = System.nanoTime();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org