Skip to content

Add table-aware GetReplicas API and deprecate legacy overloads #207

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sylwiaszunejko wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add table-aware GetReplicas API and deprecate legacy overloads #207

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions src/Cassandra.IntegrationTests/Core/MetadataTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ public void KeyspacesMetadataUpToDateViaCassandraEvents()
Assert.Greater(initialLength, 0);

//GetReplicas should yield the primary replica when the Keyspace is not found
Assert.AreEqual(1, cluster.GetReplicas("ks2", new byte[] { 0, 0, 0, 1 }).Count);
Assert.AreEqual(1, cluster.GetReplicas("ks2", null, new byte[] { 0, 0, 0, 1 }).Count);

const string createKeyspaceQuery = "CREATE KEYSPACE {0} WITH replication = {{ 'class' : '{1}', {2} }}";
session.Execute(string.Format(createKeyspaceQuery, "ks1", "NetworkTopologyStrategy", "'replication_factor' : 1"));
Expand All @@ -66,7 +66,7 @@ public void KeyspacesMetadataUpToDateViaCassandraEvents()
Assert.NotNull(ks2);
Assert.AreEqual(ks2.Replication["replication_factor"], 3);
//GetReplicas should yield the 2 replicas (rf=3 but cluster=2) when the Keyspace is found
Assert.AreEqual(2, cluster.GetReplicas("ks2", new byte[] { 0, 0, 0, 1 }).Count);
Assert.AreEqual(2, cluster.GetReplicas("ks2", null, new byte[] { 0, 0, 0, 1 }).Count);
var ks3 = cluster.Metadata.GetKeyspace("ks3");
Assert.NotNull(ks3);
Assert.AreEqual(ks3.Replication["dc1"], 1);
Expand Down
22 changes: 11 additions & 11 deletions src/Cassandra.IntegrationTests/MetadataTests/TokenMapSchemaChangeMetadataSyncTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ public void TokenMap_Should_NotUpdateExistingTokenMap_When_KeyspaceIsCreated()
newSession.ChangeKeyspace(keyspaceName);

Assert.IsNull(newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName));
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
Assert.IsTrue(object.ReferenceEquals(newCluster.Metadata.TokenToReplicasMap, oldTokenMap));
}
}
Expand All @@ -102,7 +102,7 @@ public void TokenMap_Should_NotUpdateExistingTokenMap_When_KeyspaceIsRemoved()
TestUtils.WaitForSchemaAgreement(newCluster);
var oldTokenMap = newCluster.Metadata.TokenToReplicasMap;
Assert.IsNull(newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName));
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
Assert.IsTrue(object.ReferenceEquals(newCluster.Metadata.TokenToReplicasMap, oldTokenMap));
}
}
Expand All @@ -126,7 +126,7 @@ public void TokenMap_Should_NotUpdateExistingTokenMap_When_KeyspaceIsChanged()
{
var replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.IsNull(replicas);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
});

Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Expand All @@ -139,7 +139,7 @@ public void TokenMap_Should_NotUpdateExistingTokenMap_When_KeyspaceIsChanged()
{
var replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.IsNull(replicas);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
});

Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Expand Down Expand Up @@ -169,17 +169,17 @@ public async Task TokenMap_Should_RefreshTokenMapForSingleKeyspace_When_RefreshS
var oldTokenMap = newCluster.Metadata.TokenToReplicasMap;
var replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.IsNull(replicas);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);

await newCluster.RefreshSchemaAsync(keyspaceName).ConfigureAwait(false);

Assert.AreEqual(1, newCluster.Metadata.KeyspacesSnapshot.Length);

replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.AreEqual(newCluster.Metadata.Hosts.Sum(h => h.Tokens.Count()), replicas.Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);

Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Assert.IsTrue(object.ReferenceEquals(newCluster.Metadata.TokenToReplicasMap, oldTokenMap));
Expand Down Expand Up @@ -207,22 +207,22 @@ public async Task TokenMap_Should_RefreshTokenMapForAllKeyspaces_When_RefreshSch
var newSession = newCluster.Connect(keyspaceName);
var replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.IsNull(replicas);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
var oldTokenMap = newCluster.Metadata.TokenToReplicasMap;
Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(1, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);

await newCluster.RefreshSchemaAsync().ConfigureAwait(false);

Assert.GreaterOrEqual(newCluster.Metadata.KeyspacesSnapshot.Length, 2);

replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.AreEqual(newCluster.Metadata.Hosts.Sum(h => h.Tokens.Count()), replicas.Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);

replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName1);
Assert.AreEqual(newCluster.Metadata.Hosts.Sum(h => h.Tokens.Count()), replicas.Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);

Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Assert.IsFalse(object.ReferenceEquals(newCluster.Metadata.TokenToReplicasMap, oldTokenMap));
Expand Down
4 changes: 2 additions & 2 deletions src/Cassandra.IntegrationTests/MetadataTests/TokenMapSchemaChangeTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ public void TokenMap_Should_UpdateExistingTokenMap_When_KeyspaceIsChanged()
{
var replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.AreEqual(newCluster.Metadata.Hosts.Sum(h => h.Tokens.Count()), replicas.Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(3, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
});

Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Expand All @@ -109,7 +109,7 @@ public void TokenMap_Should_UpdateExistingTokenMap_When_KeyspaceIsChanged()
{
var replicas = newCluster.Metadata.TokenToReplicasMap.GetByKeyspace(keyspaceName);
Assert.AreEqual(newCluster.Metadata.Hosts.Sum(h => h.Tokens.Count()), replicas.Count);
Assert.AreEqual(2, newCluster.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123")).Count);
Assert.AreEqual(2, newCluster.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123")).Count);
});

Assert.AreEqual(3, newCluster.Metadata.Hosts.Count(h => h.IsUp));
Expand Down
12 changes: 6 additions & 6 deletions src/Cassandra.IntegrationTests/MetadataTests/TokenMapTopologyChangeTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,8 +78,8 @@ public void TokenMap_Should_RebuildTokenMap_When_NodeIsDecommissioned()
Assert.AreEqual(3, ClusterObjSync.Metadata.Hosts.Count);
Assert.AreEqual(3, ClusterObjNotSync.Metadata.Hosts.Count);

replicasSync = ClusterObjSync.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123"));
replicasNotSync = ClusterObjNotSync.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123"));
replicasSync = ClusterObjSync.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123"));
replicasNotSync = ClusterObjNotSync.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123"));

Assert.AreEqual(3, replicasSync.Count);
Assert.AreEqual(1, replicasNotSync.Count);
Expand All @@ -97,8 +97,8 @@ public void TokenMap_Should_RebuildTokenMap_When_NodeIsDecommissioned()
Assert.AreEqual(2, ClusterObjSync.Metadata.Hosts.Count, "ClusterObjSync.Metadata.Hosts.Count");
Assert.AreEqual(2, ClusterObjNotSync.Metadata.Hosts.Count, "ClusterObjNotSync.Metadata.Hosts.Count");

replicasSync = ClusterObjSync.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123"));
replicasNotSync = ClusterObjNotSync.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123"));
replicasSync = ClusterObjSync.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123"));
replicasNotSync = ClusterObjNotSync.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123"));

Assert.AreEqual(2, replicasSync.Count, "replicasSync.Count");
Assert.AreEqual(1, replicasNotSync.Count, "replicasNotSync.Count");
Expand All @@ -116,8 +116,8 @@ public void TokenMap_Should_RebuildTokenMap_When_NodeIsDecommissioned()
Assert.AreEqual(3, ClusterObjSync.Metadata.Hosts.Count);
Assert.AreEqual(3, ClusterObjNotSync.Metadata.Hosts.Count);

replicasSync = ClusterObjSync.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123"));
replicasNotSync = ClusterObjNotSync.Metadata.GetReplicas(keyspaceName, Encoding.UTF8.GetBytes("123"));
replicasSync = ClusterObjSync.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123"));
replicasNotSync = ClusterObjNotSync.Metadata.GetReplicas(keyspaceName, null, Encoding.UTF8.GetBytes("123"));

Assert.AreEqual(3, replicasSync.Count);
Assert.AreEqual(1, replicasNotSync.Count);
Expand Down
2 changes: 1 addition & 1 deletion src/Cassandra.IntegrationTests/Policies/Tests/LoadBalancingPolicyShortTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -396,7 +396,7 @@ public void Token_Aware_Uses_Keyspace_From_Statement_To_Determine_Replication(bo
// Manually calculate the routing key
var routingKey = SerializerManager.Default.GetCurrentSerializer().Serialize(id);
// Get the replicas
var replicas = cluster.GetReplicas(ks, routingKey);
var replicas = cluster.GetReplicas(ks, "tbl1", routingKey);
Assert.AreEqual(metadataSync ? 2 : 1, replicas.Count);
CollectionAssert.AreEquivalent(replicas.Select(h => h.Host.Address), coordinators);
}
Expand Down
2 changes: 1 addition & 1 deletion src/Cassandra.IntegrationTests/ScyllaLwtTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@ private Host GetReplicaOwner()
{
var routingKey = new RoutingKey();
routingKey.RawRoutingKey = PkBytes;
var replicas = _cluster.GetReplicas("lwt_test", routingKey.RawRoutingKey);
var replicas = _cluster.GetReplicas("lwt_test", "foo", routingKey.RawRoutingKey);
return replicas.First().Host;
}

Expand Down
8 changes: 4 additions & 4 deletions src/Cassandra.Tests/PoliciesUnitTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -489,8 +489,8 @@ public void TokenAwarePolicyReturnsLocalReplicasFirst()
.Returns(hostList)
.Verifiable();
clusterMock
.Setup(c => c.GetReplicas(It.IsAny<string>(), It.IsAny<byte[]>()))
.Returns<string, byte[]>((keyspace, key) =>
.Setup(c => c.GetReplicas(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<byte[]>()))
.Returns<string, string, byte[]>((keyspace, table, key) =>
{
var i = key[0];
return hostList.Where(h =>
Expand Down Expand Up @@ -550,8 +550,8 @@ public void TokenAwarePolicyRoundRobinsOnLocalReplicas()
.Returns(hostList)
.Verifiable();
clusterMock
.Setup(c => c.GetReplicas(It.IsAny<string>(), It.IsAny<byte[]>()))
.Returns<string, byte[]>((keyspace, key) =>
.Setup(c => c.GetReplicas(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<byte[]>()))
.Returns<string, string, byte[]>((keyspace, table, key) =>
{
var i = key[0];
return hostList.Where(h =>
Expand Down
4 changes: 2 additions & 2 deletions src/Cassandra.Tests/StatementTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,7 +332,7 @@ public void BatchStatement_Should_UseRoutingKeyAndKeyspaceOfFirstStatement_When_
var rawRoutingKey = new byte[] { 1, 2, 3, 4 };
var lbp = new TokenAwarePolicy(new ClusterTests.FakeLoadBalancingPolicy());
var clusterMock = Mock.Of<IInternalCluster>();
Mock.Get(clusterMock).Setup(c => c.GetReplicas(It.IsAny<string>(), It.IsAny<byte[]>()))
Mock.Get(clusterMock).Setup(c => c.GetReplicas(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<byte[]>()))
.Returns(new List<HostShard>());
Mock.Get(clusterMock).Setup(c => c.AllHosts())
.Returns(new List<Host>());
Expand All @@ -345,7 +345,7 @@ public void BatchStatement_Should_UseRoutingKeyAndKeyspaceOfFirstStatement_When_

var _ = lbp.NewQueryPlan("ks2", batch).ToList();

Mock.Get(clusterMock).Verify(c => c.GetReplicas("ks1", rawRoutingKey), Times.Once);
Mock.Get(clusterMock).Verify(c => c.GetReplicas("ks1", null, rawRoutingKey), Times.Once);
}
}
}
12 changes: 10 additions & 2 deletions src/Cassandra/Cluster.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -490,15 +490,23 @@ public Host GetHost(IPEndPoint address)
}

/// <inheritdoc />
[Obsolete("Use GetReplicas(string keyspace, string table, byte[] partitionKey) for tablet-aware replica resolution.")]
public ICollection<HostShard> GetReplicas(byte[] partitionKey)
{
return Metadata.GetReplicas(partitionKey);
return Metadata.GetReplicas(null, null, partitionKey);
}

/// <inheritdoc />
[Obsolete("Use GetReplicas(string keyspace, string table, byte[] partitionKey) for tablet-aware replica resolution.")]
public ICollection<HostShard> GetReplicas(string keyspace, byte[] partitionKey)
{
return Metadata.GetReplicas(keyspace, partitionKey);
return Metadata.GetReplicas(keyspace, null, partitionKey);
}

/// <inheritdoc />
public ICollection<HostShard> GetReplicas(string keyspace, string table, byte[] partitionKey)
{
return Metadata.GetReplicas(keyspace, table, partitionKey);
}

private void OnHostRemoved(Host h)
Expand Down
24 changes: 18 additions & 6 deletions src/Cassandra/ICluster.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,9 @@ namespace Cassandra
/// Cassandra cluster would be:
/// </para>
/// <code>
/// Cluster cluster = Cluster.Builder().AddContactPoint("192.168.0.1").Build();
/// Session session = cluster.Connect("db1");
/// foreach (var row in session.Execute("SELECT * FROM table1"))
/// Cluster cluster = Cluster.Builder().AddContactPoint("192.168.0.1").Build();
/// Session session = cluster.Connect("db1");
/// foreach (var row in session.Execute("SELECT * FROM table1"))
/// // do something ...
/// </code>
/// <para>
Expand All @@ -46,7 +46,7 @@ namespace Cassandra
public interface ICluster : IDisposable
{
/// <summary>
/// Gets read-only metadata on the connected cluster.
/// Gets read-only metadata on the connected cluster.
/// <para>This includes the
/// know nodes (with their status as seen by the driver) as well as the schema
/// definitions.
Expand Down Expand Up @@ -108,20 +108,32 @@ public interface ICluster : IDisposable
Host GetHost(IPEndPoint address);

/// <summary>
/// Gets a collection of replicas for a given partitionKey. Backward-compatibility only, use GetReplicas(keyspace, partitionKey) instead.
/// Gets a collection of replicas for a given partitionKey. Backward-compatibility only.
/// </summary>
/// <param name="partitionKey">Byte array representing the partition key</param>
/// <returns></returns>
[Obsolete("Use GetReplicas(string keyspace, string table, byte[] partitionKey) for tablet-aware replica resolution.")]
ICollection<HostShard> GetReplicas(byte[] partitionKey);

/// <summary>
/// Gets a collection of replicas for a given partitionKey on a given keyspace
/// </summary>
/// <param name="keyspace">Byte array representing the partition key</param>
/// <param name="keyspace">The keyspace name</param>
/// <param name="partitionKey">Byte array representing the partition key</param>
/// <returns></returns>
[Obsolete("Use GetReplicas(string keyspace, string table, byte[] partitionKey) for tablet-aware replica resolution.")]
ICollection<HostShard> GetReplicas(string keyspace, byte[] partitionKey);

/// <summary>
/// Gets a collection of replicas for a given partition key on a given keyspace and table.
/// Tablet metadata is used when available, with token-map fallback otherwise.
/// </summary>
/// <param name="keyspace">The keyspace name.</param>
/// <param name="table">The table name.</param>
/// <param name="partitionKey">Byte array representing the partition key.</param>
/// <returns>A collection of replicas for the provided partition key.</returns>
ICollection<HostShard> GetReplicas(string keyspace, string table, byte[] partitionKey);

/// <summary>
/// Shutdown this cluster instance. This closes all connections from all the
/// sessions of this <c>* Cluster</c> instance and reclaim all resources
Expand Down
Loading
Loading