LOGBROKER-11013 Better fifo (#31159)

Co-authored-by: Roman Kuzin <[email protected]>

Author: kuzin57

ydb/core/persqueue/pqtablet/partition/mlp/mlp_consumer.cpp

@@ -676,7 +676,7 @@ bool TConsumerActor::FetchMessagesIfNeeded() {
         LOG_D("Skip fetch: infly limit exceeded");
         return false;
     }
-    if (metrics.InflightMessageCount >= Storage->MinMessages && metrics.UnprocessedMessageCount >= metrics.LockedMessageCount * 2) {
+    if (!Config.GetKeepMessageOrder() && metrics.InflightMessageCount >= Storage->MinMessages && metrics.UnprocessedMessageCount >= metrics.LockedMessageCount * 2) {
         LOG_D("Skip fetch: there are enough messages. InflightMessageCount=" << metrics.InflightMessageCount
             << ", UnprocessedMessageCount=" << metrics.UnprocessedMessageCount
             << ", LockedMessageCount=" << metrics.LockedMessageCount);

ydb/public/lib/ydb_cli/commands/sqs_workload/sqs_json/sqs_json_client.cpp

@@ -5,13 +5,13 @@
 #include <aws/core/http/HttpClientFactory.h>
 #include <aws/core/utils/Array.h>
 #include <aws/core/utils/HashingUtils.h>
-#include <aws/core/utils/UUID.h>
 #include <aws/core/utils/memory/stl/SimpleStringStream.h>
 #include <aws/sqs/model/DeleteMessageBatchRequest.h>
 #include <aws/sqs/model/MessageSystemAttributeNameForSends.h>
 #include <aws/sqs/model/ReceiveMessageRequest.h>
 #include <aws/sqs/model/SendMessageBatchRequest.h>
 #include <library/cpp/string_utils/url/url.h>
+#include <util/generic/guid.h>
 #include <util/generic/string.h>
 #include <util/string/cast.h>
 
@@ -119,10 +119,14 @@ namespace NYdb::NConsoleClient {
         auto credentialsProvider =
             Aws::MakeShared<Aws::Auth::SimpleAWSCredentialsProvider>(
                 "credentials-provider", credentials);
+        const Aws::String signingRegion = clientConfiguration.region.empty()
+            ? Aws::String("ru-central1")
+            : clientConfiguration.region;
+
         Signer = Aws::MakeShared<Aws::Client::AWSAuthV4Signer>(
-            "aws-auth-v4-signer", credentialsProvider, kServiceName, clientConfiguration.region,
+            "aws-auth-v4-signer", credentialsProvider, kServiceName, signingRegion,
             Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Always, true,
-            Aws::Auth::AWSSigningAlgorithm::ASYMMETRIC_SIGV4);
+            Aws::Auth::AWSSigningAlgorithm::SIGV4);
     }
 
     void TSQSJsonClient::AddHeaders(
@@ -135,8 +139,10 @@ namespace NYdb::NConsoleClient {
         request->SetHeaderValue(kContentTypeHeader, kContentTypeValue);
         request->SetHeaderValue(kAmzSdkRequestHeader, kAmzSdkRequestValue);
         request->SetHeaderValue(kXAmzAPIVersionHeader, kXAmzAPIVersionValue);
-        request->SetHeaderValue(kAmzSdkInvocationIdHeader,
-                                Aws::Utils::UUID::RandomUUID());
+        const TString invocationId = CreateGuidAsString();
+        request->SetHeaderValue(
+            kAmzSdkInvocationIdHeader,
+            Aws::String(invocationId.c_str(), invocationId.size()));
     }
 
     Aws::Utils::Json::JsonValue TSQSJsonClient::ReadResponseBody(

ydb/public/lib/ydb_cli/commands/sqs_workload/sqs_workload_writer.cpp

@@ -22,19 +22,17 @@ namespace NYdb::NConsoleClient {
         }
 
         Aws::Vector<Aws::SQS::Model::SendMessageBatchRequestEntry>
-        CreateSendMessageBatchRequestEntries(const TSqsWorkloadWriterParams& params) {
+        CreateSendMessageBatchRequestEntries(const TSqsWorkloadWriterParams& params, ui32 messageGroupID, ui32 messageDeduplicationID) {
             Aws::Vector<Aws::SQS::Model::SendMessageBatchRequestEntry> entries;
             for (ui32 i = 0; i < params.BatchSize; ++i) {
                 Aws::String messageBody(params.MessageSize, 'a');
                 Aws::SQS::Model::SendMessageBatchRequestEntry entry;
                 entry.WithMessageBody(messageBody).WithId(fmt::format("{}", i));
                 if (params.GroupsAmount > 0) {
-                    auto messageGroupID = std::rand() % params.GroupsAmount;
                     entry.WithMessageGroupId(fmt::format("{}", messageGroupID));
                 }
 
                 if (params.MaxUniqueMessages > 0) {
-                    auto messageDeduplicationID = std::rand() % params.MaxUniqueMessages;
                     entry.WithMessageDeduplicationId(std::format("{}", messageDeduplicationID));
                 }
 
@@ -70,10 +68,14 @@ namespace NYdb::NConsoleClient {
 
     void TSqsWorkloadWriter::RunLoop(const TSqsWorkloadWriterParams& params,
                                      TInstant endTime) {
+        std::mt19937_64 rng(std::random_device{}());
+        std::uniform_int_distribution<ui32> messageGroupsDistribution(0, params.GroupsAmount - 1);
+        std::uniform_int_distribution<ui32> messageDeduplicationDistribution(0, params.MaxUniqueMessages - 1);
+
         while (Now() < endTime && !params.ErrorFlag->load()) {
             Aws::SQS::Model::SendMessageBatchRequest sendMessageBatchRequest;
             sendMessageBatchRequest.SetQueueUrl(params.QueueUrl.c_str());
-            sendMessageBatchRequest.SetEntries(CreateSendMessageBatchRequestEntries(params));
+            sendMessageBatchRequest.SetEntries(CreateSendMessageBatchRequestEntries(params, messageGroupsDistribution(rng), messageDeduplicationDistribution(rng)));
             sendMessageBatchRequest.SetAdditionalCustomHeaderValue(
                 AMZ_TARGET_HEADER, SQS_TARGET_SEND_MESSAGE_BATCH);