Fix resolving queue path in AdvanceConsumer #505
Conversation
| Logger); | ||
| THashMap<int, THashMap<i64, TPartitionRowInfo>> partitionRowInfos; | ||
| try { | ||
| partitionRowInfos = CollectPartitionRowInfos( |
There was a problem hiding this comment.
I think that we generally try to avoid try-except clauses like this in our control flow. Let's make CollectPartitionRowInfos return a future to its result. It should be fairly easy to rewrite the logic inside into SelectRows + Apply.
| ReplicaToReplicatedTable_[replica] = replicatedTableInfo.Ref; | ||
| ReplicaToReplicatedTable_[replica] = TResolvedObject{ | ||
| .Path = replicatedTableInfo.Ref, | ||
| .ObjectType = replicatedTableInfo.ObjectType, |
There was a problem hiding this comment.
If we implement it like this, replicas will have a type of [Chaos]ReplicatedTable, while the [chaos]_replicated_table objects themselves will have a std::nullopt type.
Ideally, we would want to separate these types, but that all depends on how this will be used in the end.
| Logger)); | ||
|
|
||
| if (!partitionRowInfosOrError.IsOK()) { | ||
| YT_LOG_DEBUG("Failed to get partition row infos (Path: %v)", |
| if (!queueTableInfoOrError.IsOK()) { | ||
| THROW_ERROR_EXCEPTION("Failed to resolve queue path") | ||
| << queueTableInfoOrError; | ||
| const auto& consumerClusterTableMountCache = Client_->GetNativeConnection()->GetTableMountCache(); |
There was a problem hiding this comment.
Maybe this should be a helper function, since we are re-using code from PullConsumer?
yt/yt/ytlib/api/native/helpers.cpp
Outdated
| @@ -207,4 +215,25 @@ TError MakeRevivalError( | |||
|
|
|||
| //////////////////////////////////////////////////////////////////////////////// | |||
|
|
|||
| void CheckConsumerPermission( | |||
There was a problem hiding this comment.
Let's expose the context switch happening inside by returning a future? We can even chain this with .Apply.
| CheckConsumerPermission(Client_->GetNativeConnection(), consumerPath, Client_->GetOptions()); | ||
|
|
||
| auto registrationCheckResult = Client_->GetNativeConnection()->GetQueueConsumerRegistrationManager()->GetRegistration(queuePath, consumerPath); | ||
| if (!registrationCheckResult.Registration) { |
There was a problem hiding this comment.
Checking registrations should also go inside the helper, IMO, since it is also exactly the same.
| @@ -817,6 +816,8 @@ def test_multicluster_symlink_registrations(self, create_registration_table): | |||
| class TestDataApi(TestQueueConsumerApiBase, ReplicatedObjectBase, TestQueueAgentBase): | |||
| NUM_TEST_PARTITIONS = 2 | |||
|
|
|||
| NUM_REMOTE_CLUSTERS = 2 | |||
There was a problem hiding this comment.
Let's make a separate suite for multi-cluster API tests. The multi-cluster set-up/tear-down takes a while and would make all the other tests in the suite longer.
savnadya
force-pushed
the
fix-resolving-advance-consumer
branch
from
0273716
to
1c85073
Compare
|
It was merged here: cdc867f |
No description provided.