mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-29 11:02:08 +00:00
Merge pull request #60282 from ClickHouse/s3queue-fix-bug-and-flaky-test
s3queue: fix bug (also fixes flaky test_storage_s3_queue/test.py::test_shards_distributed)
This commit is contained in:
commit
6a7fef13ba
@ -699,7 +699,10 @@ void S3QueueFilesMetadata::setFileProcessedForOrderedModeImpl(
|
||||
{
|
||||
auto code = zk_client->tryMulti(requests, responses);
|
||||
if (code == Coordination::Error::ZOK)
|
||||
{
|
||||
LOG_TEST(log, "Moved file `{}` to processed", path);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/// Failed to update max processed node, retry.
|
||||
|
@ -80,6 +80,7 @@ StorageS3QueueSource::KeyWithInfoPtr StorageS3QueueSource::FileIterator::next(si
|
||||
{
|
||||
val = keys.front();
|
||||
keys.pop_front();
|
||||
chassert(idx == metadata->getProcessingIdForPath(val->key));
|
||||
}
|
||||
}
|
||||
else
|
||||
@ -103,7 +104,7 @@ StorageS3QueueSource::KeyWithInfoPtr StorageS3QueueSource::FileIterator::next(si
|
||||
LOG_TEST(log, "Putting key {} into queue of processor {} (total: {})",
|
||||
val->key, processing_id_for_key, sharded_keys.size());
|
||||
|
||||
if (auto it = sharded_keys.find(idx); it != sharded_keys.end())
|
||||
if (auto it = sharded_keys.find(processing_id_for_key); it != sharded_keys.end())
|
||||
{
|
||||
it->second.push_back(val);
|
||||
}
|
||||
@ -111,7 +112,7 @@ StorageS3QueueSource::KeyWithInfoPtr StorageS3QueueSource::FileIterator::next(si
|
||||
{
|
||||
throw Exception(ErrorCodes::LOGICAL_ERROR,
|
||||
"Processing id {} does not exist (Expected ids: {})",
|
||||
idx, fmt::join(metadata->getProcessingIdsForShard(current_shard), ", "));
|
||||
processing_id_for_key, fmt::join(metadata->getProcessingIdsForShard(current_shard), ", "));
|
||||
}
|
||||
}
|
||||
continue;
|
||||
|
Loading…
Reference in New Issue
Block a user