Skip to content

Commit c204835

Browse files
stanislav-shchetininGazizonoki
stanislav-shchetinin
authored and
Gazizonoki
committed
Fixed topic test build
1 parent 85c52c1 commit c204835

14 files changed

+118
-114
lines changed

‎src/client/persqueue_public/ut/basic_usage_ut.cpp

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -71,7 +71,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
7171
std::visit(TOverloaded {
7272
[&](TReadSessionEvent::TDataReceivedEvent& event) {
7373
for (auto& message: event.GetMessages()) {
74-
TString sourceId = message.GetMessageGroupId();
74+
std::string sourceId = message.GetMessageGroupId();
7575
ui32 seqNo = message.GetSeqNo();
7676
UNIT_ASSERT_VALUES_EQUAL(readMessageCount + 1, seqNo);
7777
++readMessageCount;
@@ -190,16 +190,16 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
190190
auto readSession = client.CreateReadSession(readSettings);
191191

192192
auto event = readSession->GetEvent(true);
193-
UNIT_ASSERT(event.Defined());
193+
UNIT_ASSERT(event.has_value());
194194

195195
auto& createPartitionStream = std::get<TReadSessionEvent::TCreatePartitionStreamEvent>(*event);
196196
createPartitionStream.Confirm();
197197

198198
UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvent(true, 0), TContractViolation);
199-
UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, Nothing(), 0), TContractViolation);
199+
UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, std::nullopt, 0), TContractViolation);
200200

201201
event = readSession->GetEvent(true, 1);
202-
UNIT_ASSERT(event.Defined());
202+
UNIT_ASSERT(event.has_value());
203203

204204
auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
205205
dataReceived.Commit();
@@ -380,7 +380,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
380380
std::visit(TOverloaded {
381381
[&](TReadSessionEvent::TDataReceivedEvent& event) {
382382
for (auto& message: event.GetMessages()) {
383-
TString sourceId = message.GetMessageGroupId();
383+
std::string sourceId = message.GetMessageGroupId();
384384
ui32 seqNo = message.GetSeqNo();
385385
UNIT_ASSERT_VALUES_EQUAL(readMessageCount + 1, seqNo);
386386
++readMessageCount;
@@ -447,7 +447,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
447447
public:
448448
TBrokenCredentialsProvider() {}
449449
virtual ~TBrokenCredentialsProvider() {}
450-
TStringType GetAuthInfo() const {
450+
std::string GetAuthInfo() const {
451451
ythrow yexception() << "exception during creation";
452452
return "";
453453
}
@@ -463,7 +463,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
463463
return std::make_shared<TBrokenCredentialsProvider>();
464464
}
465465

466-
virtual TStringType GetClientIdentity() const {
466+
virtual std::string GetClientIdentity() const {
467467
return "abacaba";
468468
}
469469
};

‎src/client/persqueue_public/ut/read_session_ut.cpp

Lines changed: 33 additions & 33 deletions
Large diffs are not rendered by default.

‎src/client/persqueue_public/ut/retry_policy_ut.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -316,7 +316,7 @@ Y_UNIT_TEST_SUITE(RetryPolicy) {
316316
Cerr << "===Data event\n";
317317
auto& clusterName = event.GetPartitionStream()->GetCluster();
318318
for (auto& message: event.GetMessages()) {
319-
TString sourceId = message.GetMessageGroupId();
319+
auto sourceId = TString{message.GetMessageGroupId()};
320320
ui32 seqNo = message.GetSeqNo();
321321
if (sourceId == sourceId1) {
322322
UNIT_ASSERT_VALUES_EQUAL(seqNo, seqNoByClusterSrc1[clusterName] + 1);

‎src/client/persqueue_public/ut/ut_utils/data_plane_helpers.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ namespace NKikimr::NPersQueueTests {
5151
std::optional<ui32> partitionGroup,
5252
std::optional<TString> codec,
5353
std::optional<bool> reconnectOnFailure,
54-
THashMap<TString, TString> sessionMeta
54+
std::unordered_map<std::string, std::string> sessionMeta
5555
) {
5656
auto settings = TWriteSessionSettings().Path(topic).MessageGroupId(sourceId);
5757
if (partitionGroup) settings.PartitionGroupId(*partitionGroup);
@@ -84,7 +84,7 @@ namespace NKikimr::NPersQueueTests {
8484
auto future = reader->WaitEvent();
8585
future.Wait(timeout);
8686

87-
TMaybe<NYdb::NPersQueue::TReadSessionEvent::TEvent> event = reader->GetEvent(false, 1);
87+
std::optional<NYdb::NPersQueue::TReadSessionEvent::TEvent> event = reader->GetEvent(false, 1);
8888
if (!event)
8989
return {};
9090
if (auto dataEvent = std::get_if<NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent>(&*event)) {

‎src/client/persqueue_public/ut/ut_utils/data_plane_helpers.h

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44
#include <ydb-cpp-sdk/client/table/table.h>
55
#include <src/client/persqueue_public/persqueue.h>
66

7+
#include <util/generic/hash.h>
8+
79
namespace NKikimr::NPersQueueTests {
810

911
std::shared_ptr<NYdb::NPersQueue::IWriteSession> CreateWriter(
@@ -34,7 +36,7 @@ namespace NKikimr::NPersQueueTests {
3436
std::optional<ui32> partitionGroup = {},
3537
std::optional<TString> codec = {},
3638
std::optional<bool> reconnectOnFailure = {},
37-
THashMap<TString, TString> sessionMeta = {}
39+
std::unordered_map<std::string, std::string> sessionMeta = {}
3840
);
3941

4042
std::shared_ptr<NYdb::NPersQueue::IReadSession> CreateReader(

‎src/client/persqueue_public/ut/ut_utils/sdk_test_setup.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ class SDKTestSetup {
8888
}
8989
}
9090

91-
static TString GetTestTopic() {
91+
static std::string GetTestTopic() {
9292
return "test-topic";
9393
}
9494

‎src/client/persqueue_public/ut/ut_utils/ut_utils.h

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
4545
NYdb::TDriverConfig cfg;
4646
cfg.SetEndpoint(TStringBuilder() << "localhost:" << Server.GrpcPort);
4747
cfg.SetDatabase("/Root");
48-
cfg.SetLog(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG));
48+
cfg.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG).Release()));
4949
Driver = MakeHolder<NYdb::TDriver>(cfg);
5050
}
5151
return *Driver;
@@ -168,7 +168,7 @@ struct TYDBClientEventLoop : public ::NPersQueue::IClientEventLoop {
168168
}
169169
Y_ABORT_UNLESS(continueToken);
170170

171-
TMaybe<ui64> seqNo = Nothing();
171+
std::optional<ui64> seqNo = std::nullopt;
172172
if (!AutoSeqNo) {
173173
seqNo = acknowledgeableMessage.SequenceNumber;
174174
Log << TLOG_INFO << "[" << sourceId << "] Write messages with sequence numbers "
@@ -377,7 +377,7 @@ class TYdbPqTestExecutor : public IAsyncExecutor {
377377
Stop = true;
378378
Thread.Join();
379379
}
380-
void PostImpl(TVector<TFunction>&& fs) override {
380+
void PostImpl(std::vector<TFunction>&& fs) override {
381381
for (auto& f : fs) {
382382
TasksQueue.Enqueue(std::move(f));
383383
}

‎src/client/topic/impl/common.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -381,7 +381,7 @@ class TBaseSessionEventsQueue : public ISignalable {
381381

382382
void WaitEventsImpl() { // Assumes that we're under lock. Posteffect: HasEventsImpl() is true.
383383
while (!HasEventsImpl()) {
384-
std::unique_lock<std::mutex> lk(Mutex, std::defer_lock);
384+
std::unique_lock<std::mutex> lk(Mutex, std::adopt_lock);
385385
CondVar.wait(lk);
386386
}
387387
}

‎src/client/topic/ut/basic_usage_ut.cpp

Lines changed: 21 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616
#include <library/cpp/threading/future/future.h>
1717
#include <library/cpp/threading/future/async.h>
1818

19+
#include <util/stream/zlib.h>
20+
1921
#include <future>
2022

2123
namespace NYdb::NTopic::NTests {
@@ -99,7 +101,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
99101
NYdb::TDriverConfig cfg;
100102
cfg.SetEndpoint(TStringBuilder() << "invalid:" << setup.GetServer().GrpcPort);
101103
cfg.SetDatabase("/Invalid");
102-
cfg.SetLog(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG));
104+
cfg.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG).Release()));
103105
auto driver = NYdb::TDriver(cfg);
104106

105107
{
@@ -113,13 +115,13 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
113115
auto writeSession = client.CreateWriteSession(writeSettings);
114116

115117
auto event = writeSession->GetEvent(true);
116-
UNIT_ASSERT(event.Defined() && std::holds_alternative<TSessionClosedEvent>(event.GetRef()));
118+
UNIT_ASSERT(event && std::holds_alternative<TSessionClosedEvent>(event.value()));
117119
}
118120

119121
{
120122
auto settings = TTopicClientSettings()
121123
.Database({"/Root"})
122-
.DiscoveryEndpoint({TStringBuilder() << "localhost:" << setup.GetServer().GrpcPort});
124+
.DiscoveryEndpoint("localhost:" + std::to_string(setup.GetServer().GrpcPort));
123125

124126
TTopicClient client(driver, settings);
125127

@@ -130,7 +132,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
130132
auto writeSession = client.CreateWriteSession(writeSettings);
131133

132134
auto event = writeSession->GetEvent(true);
133-
UNIT_ASSERT(event.Defined() && !std::holds_alternative<TSessionClosedEvent>(event.GetRef()));
135+
UNIT_ASSERT(event && !std::holds_alternative<TSessionClosedEvent>(event.value()));
134136
}
135137
}
136138

@@ -171,13 +173,13 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
171173
auto readSession = client.CreateReadSession(readSettings);
172174

173175
auto event = readSession->GetEvent(true);
174-
UNIT_ASSERT(event.Defined());
176+
UNIT_ASSERT(event.has_value());
175177

176178
auto& startPartitionSession = std::get<TReadSessionEvent::TStartPartitionSessionEvent>(*event);
177179
startPartitionSession.Confirm();
178180

179181
event = readSession->GetEvent(true);
180-
UNIT_ASSERT(event.Defined());
182+
UNIT_ASSERT(event.has_value());
181183

182184
auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
183185
dataReceived.Commit();
@@ -234,16 +236,16 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
234236
auto readSession = client.CreateReadSession(readSettings);
235237

236238
auto event = readSession->GetEvent(true);
237-
UNIT_ASSERT(event.Defined());
239+
UNIT_ASSERT(event.has_value());
238240

239241
auto& startPartitionSession = std::get<TReadSessionEvent::TStartPartitionSessionEvent>(*event);
240242
startPartitionSession.Confirm();
241243

242244
UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvent(true, 0), TContractViolation);
243-
UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, Nothing(), 0), TContractViolation);
245+
UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, std::nullopt, 0), TContractViolation);
244246

245247
event = readSession->GetEvent(true, 1);
246-
UNIT_ASSERT(event.Defined());
248+
UNIT_ASSERT(event.has_value());
247249

248250
auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
249251
dataReceived.Commit();
@@ -335,7 +337,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
335337
auto description = result.GetConsumerDescription();
336338
UNIT_ASSERT(description.GetPartitions().size() == 1);
337339
auto stats = description.GetPartitions().front().GetPartitionConsumerStats();
338-
UNIT_ASSERT(stats.Defined());
340+
UNIT_ASSERT(stats.has_value());
339341
UNIT_ASSERT(stats->GetCommittedOffset() == 50);
340342
}
341343

@@ -695,7 +697,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
695697
auto description = result.GetTopicDescription();
696698
UNIT_ASSERT(description.GetPartitions().size() == 1);
697699
auto stats = description.GetPartitions().front().GetPartitionStats();
698-
UNIT_ASSERT(stats.Defined());
700+
UNIT_ASSERT(stats.has_value());
699701
UNIT_ASSERT_VALUES_EQUAL(stats->GetEndOffset(), count);
700702

701703
}
@@ -779,7 +781,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
779781
std::visit(TOverloaded {
780782
[&](TReadSessionEvent::TDataReceivedEvent& event) {
781783
for (auto& message: event.GetMessages()) {
782-
TString sourceId = message.GetMessageGroupId();
784+
std::string sourceId = message.GetMessageGroupId();
783785
ui32 seqNo = message.GetSeqNo();
784786
UNIT_ASSERT_VALUES_EQUAL(readMessageCount + 1, seqNo);
785787
++readMessageCount;
@@ -829,11 +831,11 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
829831

830832
auto client = setup.MakeClient();
831833
ui64 producerIndex = 0u;
832-
auto runTest = [&](TString producer, TString msgGroup, const TMaybe<bool>& useDedup, bool useSeqNo, EExpectedTestResult result) ->bool
834+
auto runTest = [&](TString producer, TString msgGroup, const std::optional<bool>& useDedup, bool useSeqNo, EExpectedTestResult result) ->bool
833835
{
834836
TWriteSessionSettings writeSettings;
835837
writeSettings.Path(setup.GetTopicPath()).Codec(NTopic::ECodec::RAW);
836-
TString useDedupStr = useDedup.Defined() ? ToString(*useDedup) : "<unset>";
838+
TString useDedupStr = useDedup.has_value() ? ToString(*useDedup) : "<unset>";
837839
if (producer) {
838840
producer += ToString(producerIndex);
839841
}
@@ -848,7 +850,7 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
848850
<< useDedupStr << ", manual SeqNo: " << useSeqNo << Endl;
849851

850852
try {
851-
if (useDedup.Defined()) {
853+
if (useDedup.has_value()) {
852854
writeSettings.DeduplicationEnabled(useDedup);
853855
}
854856
auto session = client.CreateWriteSession(writeSettings);
@@ -857,12 +859,12 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
857859
ui64 written = 0;
858860
while (written < 10) {
859861
auto event = session->GetEvent(true);
860-
if (std::holds_alternative<TSessionClosedEvent>(event.GetRef())) {
862+
if (std::holds_alternative<TSessionClosedEvent>(event.value())) {
861863
auto closed = std::get<TSessionClosedEvent>(*event);
862864
Cerr << "Session failed with error: " << closed.DebugString() << Endl;
863865
UNIT_ASSERT(result == EExpectedTestResult::FAIL_ON_RPC);
864866
return false;
865-
} else if (std::holds_alternative<TWriteSessionEvent::TReadyToAcceptEvent>(event.GetRef())) {
867+
} else if (std::holds_alternative<TWriteSessionEvent::TReadyToAcceptEvent>(event.value())) {
866868
token = std::move(std::get<TWriteSessionEvent::TReadyToAcceptEvent>(*event).ContinuationToken);
867869
if (useSeqNo) {
868870
session->Write(std::move(*token), "data", seqNo++);
@@ -948,10 +950,10 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
948950

949951
auto readSession = client.CreateReadSession(readSettings);
950952
auto event = readSession->GetEvent(true);
951-
UNIT_ASSERT(event.Defined());
953+
UNIT_ASSERT(event.has_value());
952954

953955
auto& closeEvent = std::get<NYdb::NTopic::TSessionClosedEvent>(*event);
954-
UNIT_ASSERT(closeEvent.DebugString().Contains("Too small max memory usage"));
956+
UNIT_ASSERT(closeEvent.DebugString().contains("Too small max memory usage"));
955957
}
956958

957959
} // Y_UNIT_TEST_SUITE(TSettingsValidation)

‎src/client/topic/ut/describe_topic_ut.cpp

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -144,7 +144,7 @@ namespace NYdb::NTopic::NTests {
144144

145145
UNIT_ASSERT_GT(consumerStats->GetLastReadOffset(), 0);
146146
UNIT_ASSERT_GT(consumerStats->GetCommittedOffset(), 0);
147-
UNIT_ASSERT_GE(consumerStats->GetReadSessionId(), 0);
147+
UNIT_ASSERT_GE(TString{consumerStats->GetReadSessionId()}, 0);
148148
UNIT_ASSERT_VALUES_EQUAL(consumerStats->GetReaderName(), "");
149149
} else {
150150
UNIT_ASSERT_VALUES_EQUAL(stats->GetStartOffset(), 0);
@@ -296,29 +296,29 @@ namespace NYdb::NTopic::NTests {
296296

297297
// Event 1: start partition session
298298
{
299-
TMaybe<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
299+
std::optional<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
300300
UNIT_ASSERT(event);
301-
auto startPartitionSession = std::get_if<TReadSessionEvent::TStartPartitionSessionEvent>(event.Get());
301+
auto startPartitionSession = std::get_if<TReadSessionEvent::TStartPartitionSessionEvent>(&event.value());
302302
UNIT_ASSERT_C(startPartitionSession, DebugString(*event));
303303

304304
startPartitionSession->Confirm();
305305
}
306306

307307
// Event 2: data received
308308
{
309-
TMaybe<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
309+
std::optional<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
310310
UNIT_ASSERT(event);
311-
auto dataReceived = std::get_if<TReadSessionEvent::TDataReceivedEvent>(event.Get());
311+
auto dataReceived = std::get_if<TReadSessionEvent::TDataReceivedEvent>(&event.value());
312312
UNIT_ASSERT_C(dataReceived, DebugString(*event));
313313

314314
dataReceived->Commit();
315315
}
316316

317317
// Event 3: commit acknowledgement
318318
{
319-
TMaybe<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
319+
std::optional<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
320320
UNIT_ASSERT(event);
321-
auto commitOffsetAck = std::get_if<TReadSessionEvent::TCommitOffsetAcknowledgementEvent>(event.Get());
321+
auto commitOffsetAck = std::get_if<TReadSessionEvent::TCommitOffsetAcknowledgementEvent>(&event.value());
322322

323323
UNIT_ASSERT_C(commitOffsetAck, DebugString(*event));
324324

@@ -363,7 +363,7 @@ namespace NYdb::NTopic::NTests {
363363
if (allowUpdateRow) {
364364
acl.AddAccess(NACLib::EAccessType::Allow, NACLib::UpdateRow, authToken);
365365
}
366-
setup.GetServer().AnnoyingClient->ModifyACL("/Root", TEST_TOPIC, acl.SerializeAsString());
366+
setup.GetServer().AnnoyingClient->ModifyACL("/Root", TString{TEST_TOPIC}, acl.SerializeAsString());
367367

368368
return client.DescribePartition(existingTopic ? TEST_TOPIC : "bad-topic", testPartitionId, settings).GetValueSync();
369369
}
@@ -409,7 +409,7 @@ namespace NYdb::NTopic::NTests {
409409
if (resultStatus == EStatus::SUCCESS) {
410410
auto& p = result.GetPartitionDescription().GetPartition();
411411
UNIT_ASSERT(p.GetActive());
412-
UNIT_ASSERT(p.GetPartitionLocation().Defined());
412+
UNIT_ASSERT(p.GetPartitionLocation().has_value());
413413
}
414414
}
415415
}

0 commit comments

Comments
 (0)