Skip to content

Commit a6efcba

Browse files
pnv1pnv1
authored
Merge pull request #48 from ydb-platform/update-transaction-topic-examples
Update transaction topic examples
2 parents 39c84c8 + 530deb4 commit a6efcba

File tree

18 files changed

+285
-418
lines changed

18 files changed

+285
-418
lines changed

‎jdbc/spring-jooq/pom.xml‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -106,7 +106,6 @@
106106
<configuration>
107107
<environmentVariables>
108108
<TESTCONTAINERS_REUSE_ENABLE>true</TESTCONTAINERS_REUSE_ENABLE>
109-
<YDB_DOCKER_IMAGE>cr.yandex/yc/yandex-docker-local-ydb:trunk</YDB_DOCKER_IMAGE>
110109
</environmentVariables>
111110
</configuration>
112111
</plugin>
@@ -211,4 +210,4 @@
211210
</plugin>
212211
</plugins>
213212
</build>
214-
</project>
213+
</project>

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
<log4j.version>2.22.1</log4j.version>
1919
<jcommander.version>1.82</jcommander.version>
2020

21-
<ydb.sdk.version>2.2.10</ydb.sdk.version>
21+
<ydb.sdk.version>2.3.10</ydb.sdk.version>
2222
</properties>
2323

2424
<modules>

‎query-example/pom.xml‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,6 @@
7373
<artifactId>maven-surefire-plugin</artifactId>
7474
<configuration>
7575
<environmentVariables>
76-
<YDB_DOCKER_IMAGE>cr.yandex/yc/yandex-docker-local-ydb:trunk</YDB_DOCKER_IMAGE>
7776
<TESTCONTAINERS_REUSE_ENABLE>true</TESTCONTAINERS_REUSE_ENABLE>
7877
<YDB_ANONYMOUS_CREDENTIALS>1</YDB_ANONYMOUS_CREDENTIALS>
7978
</environmentVariables>

‎query-example/src/main/java/tech/ydb/example/App.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ private void upsertTablesData() {
119119

120120
// Upsert list of series to table
121121
retryCtx.supplyResult(session -> session.createQuery(
122+
"DECLARE $values AS " + ListType.of(seriesType) + ";" +
122123
"UPSERT INTO series SELECT * FROM AS_TABLE($values)",
123124
TxMode.SERIALIZABLE_RW,
124125
Params.of("$values", seriesData)
@@ -146,6 +147,7 @@ private void upsertTablesData() {
146147

147148
// Upsert list of seasons to table
148149
retryCtx.supplyResult(session -> session.createQuery(
150+
"DECLARE $values AS " + ListType.of(seasonType) + ";" +
149151
"UPSERT INTO seasons SELECT * FROM AS_TABLE($values)",
150152
TxMode.SERIALIZABLE_RW,
151153
Params.of("$values", seasonsData)
@@ -173,6 +175,7 @@ private void upsertTablesData() {
173175

174176
// Upsert list of series to episodes
175177
retryCtx.supplyResult(session -> session.createQuery(
178+
"DECLARE $values AS " + ListType.of(episodeType) + ";" +
176179
"UPSERT INTO episodes SELECT * FROM AS_TABLE($values)",
177180
TxMode.SERIALIZABLE_RW,
178181
Params.of("$values", episodesData)

‎ydb-cookbook/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,10 @@
2626
<groupId>tech.ydb</groupId>
2727
<artifactId>ydb-sdk-scheme</artifactId>
2828
</dependency>
29+
<dependency>
30+
<groupId>tech.ydb</groupId>
31+
<artifactId>ydb-sdk-query</artifactId>
32+
</dependency>
2933
<dependency>
3034
<groupId>tech.ydb</groupId>
3135
<artifactId>ydb-sdk-topic</artifactId>

‎ydb-cookbook/src/main/java/tech/ydb/examples/SimpleExample.java‎

Lines changed: 4 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,14 @@
11
package tech.ydb.examples;
22

3-
import java.time.Duration;
4-
53
import tech.ydb.auth.iam.CloudAuthHelper;
6-
import tech.ydb.core.Result;
7-
import tech.ydb.core.Status;
84
import tech.ydb.core.grpc.GrpcTransport;
9-
import tech.ydb.scheme.SchemeClient;
10-
import tech.ydb.scheme.description.DescribePathResult;
11-
import tech.ydb.topic.TopicClient;
12-
import tech.ydb.topic.description.Codec;
13-
import tech.ydb.topic.description.Consumer;
14-
import tech.ydb.topic.description.SupportedCodecs;
15-
import tech.ydb.topic.settings.CreateTopicSettings;
165

176

187
/**
198
* @author Sergey Polovko
209
* @author Nikolay Perfilov
2110
*/
2211
public abstract class SimpleExample {
23-
protected static final String TOPIC_NAME = System.getenv("YDB_TOPIC_NAME");
24-
protected static final String CONSUMER_NAME = System.getenv("YDB_CONSUMER_NAME");
25-
2612
protected void doMain(String[] args) {
2713
if (args.length > 1) {
2814
System.err.println("Too many arguments");
@@ -32,11 +18,12 @@ protected void doMain(String[] args) {
3218
if (args.length == 1) {
3319
connString = args[0];
3420
} else {
35-
connString = "some.host.name.com:2135?database=/Root";
36-
System.err.println("Pass <connection-string> as argument to override connection settings\n");
21+
System.err.println("Pass <connection-string> as an argument. " +
22+
"Example: some.host.name.com:2135?database=/Root\n");
23+
return;
3724
}
3825

39-
System.err.println("connection-string: " + connString + "\n");
26+
System.out.println("connection-string: " + connString + "\n");
4027

4128
try (GrpcTransport transport = GrpcTransport.forConnectionString(connString)
4229
.withAuthProvider(CloudAuthHelper.getAuthProviderFromEnviron())

‎ydb-cookbook/src/main/java/tech/ydb/examples/topic/ControlPlane.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@
1212
import org.slf4j.LoggerFactory;
1313
import tech.ydb.core.Result;
1414
import tech.ydb.core.grpc.GrpcTransport;
15-
import tech.ydb.examples.SimpleExample;
1615
import tech.ydb.topic.TopicClient;
1716
import tech.ydb.topic.description.Codec;
1817
import tech.ydb.topic.description.Consumer;
@@ -28,11 +27,11 @@
2827
/**
2928
* @author Nikolay Perfilov
3029
*/
31-
public class ControlPlane extends SimpleExample {
30+
public class ControlPlane extends SimpleTopicExample {
3231
private static final Logger logger = LoggerFactory.getLogger(ControlPlane.class);
3332

3433
@Override
35-
protected void run(GrpcTransport transport, String pathPrefix) {
34+
protected void run(GrpcTransport transport) {
3635
logger.info("ControlPlane run");
3736
try (TopicClient topicClient = TopicClient.newClient(transport).build()) {
3837

‎ydb-cookbook/src/main/java/tech/ydb/examples/topic/ReadAsync.java‎

Lines changed: 2 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@
99
import org.slf4j.Logger;
1010
import org.slf4j.LoggerFactory;
1111
import tech.ydb.core.grpc.GrpcTransport;
12-
import tech.ydb.examples.SimpleExample;
1312
import tech.ydb.topic.TopicClient;
1413
import tech.ydb.topic.read.AsyncReader;
1514
import tech.ydb.topic.read.DecompressionException;
@@ -28,16 +27,15 @@
2827
/**
2928
* @author Nikolay Perfilov
3029
*/
31-
public class ReadAsync extends SimpleExample {
30+
public class ReadAsync extends SimpleTopicExample {
3231
private static final Logger logger = LoggerFactory.getLogger(ReadAsync.class);
3332
private static final long MAX_MEMORY_USAGE_BYTES = 500 * 1024 * 1024; // 500 Mb
3433
private static final int MESSAGES_COUNT = 5;
3534

3635
private final CompletableFuture<Void> messageReceivedFuture = new CompletableFuture<>();
37-
private long lastSeqNo = -1;
3836

3937
@Override
40-
protected void run(GrpcTransport transport, String pathPrefix) {
38+
protected void run(GrpcTransport transport) {
4139

4240
try (TopicClient topicClient = TopicClient.newClient(transport)
4341
.setCompressionPoolThreadCount(8)
@@ -108,13 +106,6 @@ public void onMessages(DataReceivedEvent event) {
108106
} else {
109107
logger.info("Message received. SeqNo={}, offset={}", message.getSeqNo(), message.getOffset());
110108
}
111-
if (lastSeqNo > message.getSeqNo()) {
112-
logger.error("Received a message with seqNo {}. Previously got a message with seqNo {}",
113-
message.getSeqNo(), lastSeqNo);
114-
messageReceivedFuture.complete(null);
115-
} else {
116-
lastSeqNo = message.getSeqNo();
117-
}
118109
message.commit().thenRun(() -> {
119110
logger.info("Message committed");
120111
if (messageCounter.incrementAndGet() >= MESSAGES_COUNT) {

‎ydb-cookbook/src/main/java/tech/ydb/examples/topic/ReadSync.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
import org.slf4j.Logger;
88
import org.slf4j.LoggerFactory;
99
import tech.ydb.core.grpc.GrpcTransport;
10-
import tech.ydb.examples.SimpleExample;
1110
import tech.ydb.topic.TopicClient;
1211
import tech.ydb.topic.read.DecompressionException;
1312
import tech.ydb.topic.read.Message;
@@ -18,11 +17,11 @@
1817
/**
1918
* @author Nikolay Perfilov
2019
*/
21-
public class ReadSync extends SimpleExample {
20+
public class ReadSync extends SimpleTopicExample {
2221
private static final Logger logger = LoggerFactory.getLogger(ReadSync.class);
2322

2423
@Override
25-
protected void run(GrpcTransport transport, String pathPrefix) {
24+
protected void run(GrpcTransport transport) {
2625

2726
try (TopicClient topicClient = TopicClient.newClient(transport).build()) {
2827

‎ydb-cookbook/src/main/java/tech/ydb/examples/topic/ReadWriteWorkload.java‎

Lines changed: 3 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
import org.slf4j.Logger;
2020
import org.slf4j.LoggerFactory;
2121
import tech.ydb.core.grpc.GrpcTransport;
22-
import tech.ydb.examples.SimpleExample;
2322
import tech.ydb.topic.TopicClient;
2423
import tech.ydb.topic.description.Codec;
2524
import tech.ydb.topic.read.AsyncReader;
@@ -40,7 +39,7 @@
4039
/**
4140
* @author Nikolay Perfilov
4241
*/
43-
public class ReadWriteWorkload extends SimpleExample {
42+
public class ReadWriteWorkload extends SimpleTopicExample {
4443
private static final Logger logger = LoggerFactory.getLogger(ReadWriteWorkload.class);
4544
private static final int WRITE_TIMEOUT_SECONDS = 60;
4645
private static final int MESSAGE_LENGTH_BYTES = 10_000_000; // 10 Mb
@@ -53,12 +52,11 @@ public class ReadWriteWorkload extends SimpleExample {
5352
private final AtomicInteger messagesReceived = new AtomicInteger(0);
5453
private final AtomicInteger messagesCommitted = new AtomicInteger(0);
5554
private final AtomicLong bytesWritten = new AtomicLong(0);
56-
private long lastSeqNo = -1;
5755
CountDownLatch writeFinishedLatch = new CountDownLatch(1);
5856
CountDownLatch readFinishedLatch = new CountDownLatch(1);
5957

6058
@Override
61-
protected void run(GrpcTransport transport, String pathPrefix) {
59+
protected void run(GrpcTransport transport) {
6260

6361
ExecutorService compressionExecutor = Executors.newFixedThreadPool(10);
6462
AtomicBoolean timeToStopWriting = new AtomicBoolean(false);
@@ -154,9 +152,8 @@ protected void run(GrpcTransport transport, String pathPrefix) {
154152
};
155153

156154
Runnable readingThread = () -> {
157-
String consumerName = "consumer1";
158155
ReaderSettings readerSettings = ReaderSettings.newBuilder()
159-
.setConsumerName(consumerName)
156+
.setConsumerName(CONSUMER_NAME)
160157
.addTopic(TopicReadSettings.newBuilder()
161158
.setPath(TOPIC_NAME)
162159
.setReadFrom(Instant.now().minus(Duration.ofHours(24)))
@@ -274,12 +271,6 @@ public void onMessages(DataReceivedEvent event) {
274271
} else {
275272
logger.debug("Message received. SeqNo={}, offset={}", message.getSeqNo(), message.getOffset());
276273
}
277-
if (lastSeqNo > message.getSeqNo()) {
278-
logger.error("Received a message with seqNo {}. Previously got a message with seqNo {}",
279-
message.getSeqNo(), lastSeqNo);
280-
} else {
281-
lastSeqNo = message.getSeqNo();
282-
}
283274
message.commit().thenRun(() -> {
284275
logger.trace("Message committed");
285276
unreadMessagesCount.decrementAndGet();

0 commit comments

Comments
 (0)