Update transaction topic examples

jdbc/spring-jooq/pom.xml

@@ -106,7 +106,6 @@
                 <configuration>
                     <environmentVariables>
                         <TESTCONTAINERS_REUSE_ENABLE>true</TESTCONTAINERS_REUSE_ENABLE>
-                        <YDB_DOCKER_IMAGE>cr.yandex/yc/yandex-docker-local-ydb:trunk</YDB_DOCKER_IMAGE>
                     </environmentVariables>
                 </configuration>
             </plugin>
@@ -210,4 +209,4 @@
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

pom.xml

@@ -18,7 +18,7 @@
         <log4j.version>2.22.1</log4j.version>
         <jcommander.version>1.82</jcommander.version>
 
-        <ydb.sdk.version>2.2.10</ydb.sdk.version>
+        <ydb.sdk.version>2.3.10</ydb.sdk.version>
     </properties>
 
     <modules>

query-example/pom.xml

@@ -73,7 +73,6 @@
                 <artifactId>maven-surefire-plugin</artifactId>
                 <configuration>
                     <environmentVariables>
-                        <YDB_DOCKER_IMAGE>cr.yandex/yc/yandex-docker-local-ydb:trunk</YDB_DOCKER_IMAGE>
                         <TESTCONTAINERS_REUSE_ENABLE>true</TESTCONTAINERS_REUSE_ENABLE>
                         <YDB_ANONYMOUS_CREDENTIALS>1</YDB_ANONYMOUS_CREDENTIALS>
                     </environmentVariables>

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

@@ -119,6 +119,7 @@ private void upsertTablesData() {
 
         // Upsert list of series to table
         retryCtx.supplyResult(session -> session.createQuery(
+                "DECLARE $values AS " + ListType.of(seriesType) + ";" +
                 "UPSERT INTO series SELECT * FROM AS_TABLE($values)",
                 TxMode.SERIALIZABLE_RW,
                 Params.of("$values", seriesData)
@@ -146,6 +147,7 @@ private void upsertTablesData() {
 
         // Upsert list of seasons to table
         retryCtx.supplyResult(session -> session.createQuery(
+                "DECLARE $values AS " + ListType.of(seasonType) + ";" +
                 "UPSERT INTO seasons SELECT * FROM AS_TABLE($values)",
                 TxMode.SERIALIZABLE_RW,
                 Params.of("$values", seasonsData)
@@ -173,6 +175,7 @@ private void upsertTablesData() {
 
         // Upsert list of series to episodes
         retryCtx.supplyResult(session -> session.createQuery(
+                "DECLARE $values AS " + ListType.of(episodeType) + ";" +
                 "UPSERT INTO episodes SELECT * FROM AS_TABLE($values)",
                 TxMode.SERIALIZABLE_RW,
                 Params.of("$values", episodesData)

ydb-cookbook/pom.xml

@@ -26,6 +26,10 @@
             <groupId>tech.ydb</groupId>
             <artifactId>ydb-sdk-scheme</artifactId>
         </dependency>
+        <dependency>
+            <groupId>tech.ydb</groupId>
+            <artifactId>ydb-sdk-query</artifactId>
+        </dependency>
         <dependency>
             <groupId>tech.ydb</groupId>
             <artifactId>ydb-sdk-topic</artifactId>

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

@@ -1,28 +1,14 @@
 package tech.ydb.examples;
 
-import java.time.Duration;
-
 import tech.ydb.auth.iam.CloudAuthHelper;
-import tech.ydb.core.Result;
-import tech.ydb.core.Status;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.scheme.SchemeClient;
-import tech.ydb.scheme.description.DescribePathResult;
-import tech.ydb.topic.TopicClient;
-import tech.ydb.topic.description.Codec;
-import tech.ydb.topic.description.Consumer;
-import tech.ydb.topic.description.SupportedCodecs;
-import tech.ydb.topic.settings.CreateTopicSettings;
 
 
 /**
  * @author Sergey Polovko
  * @author Nikolay Perfilov
  */
 public abstract class SimpleExample {
-    protected static final String TOPIC_NAME = System.getenv("YDB_TOPIC_NAME");
-    protected static final String CONSUMER_NAME = System.getenv("YDB_CONSUMER_NAME");
-
     protected void doMain(String[] args) {
         if (args.length > 1) {
             System.err.println("Too many arguments");
@@ -32,11 +18,12 @@ protected void doMain(String[] args) {
         if (args.length == 1) {
             connString = args[0];
         } else {
-            connString = "some.host.name.com:2135?database=/Root";
-            System.err.println("Pass <connection-string> as argument to override connection settings\n");
+            System.err.println("Pass <connection-string> as an argument. " +
+                    "Example: some.host.name.com:2135?database=/Root\n");
+            return;
         }
 
-        System.err.println("connection-string: " + connString + "\n");
+        System.out.println("connection-string: " + connString + "\n");
 
         try (GrpcTransport transport = GrpcTransport.forConnectionString(connString)
                 .withAuthProvider(CloudAuthHelper.getAuthProviderFromEnviron())

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

@@ -12,7 +12,6 @@
 import org.slf4j.LoggerFactory;
 import tech.ydb.core.Result;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.examples.SimpleExample;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.description.Consumer;
@@ -28,11 +27,11 @@
 /**
  * @author Nikolay Perfilov
  */
-public class ControlPlane extends SimpleExample {
+public class ControlPlane extends SimpleTopicExample {
     private static final Logger logger = LoggerFactory.getLogger(ControlPlane.class);
 
     @Override
-    protected void run(GrpcTransport transport, String pathPrefix) {
+    protected void run(GrpcTransport transport) {
         logger.info("ControlPlane run");
         try (TopicClient topicClient = TopicClient.newClient(transport).build()) {
 

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

@@ -9,7 +9,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.examples.SimpleExample;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.read.AsyncReader;
 import tech.ydb.topic.read.DecompressionException;
@@ -28,16 +27,15 @@
 /**
  * @author Nikolay Perfilov
  */
-public class ReadAsync extends SimpleExample {
+public class ReadAsync extends SimpleTopicExample {
     private static final Logger logger = LoggerFactory.getLogger(ReadAsync.class);
     private static final long MAX_MEMORY_USAGE_BYTES = 500 * 1024 * 1024; // 500 Mb
     private static final int MESSAGES_COUNT = 5;
 
     private final CompletableFuture<Void> messageReceivedFuture = new CompletableFuture<>();
-    private long lastSeqNo = -1;
 
     @Override
-    protected void run(GrpcTransport transport, String pathPrefix) {
+    protected void run(GrpcTransport transport) {
 
         try (TopicClient topicClient = TopicClient.newClient(transport)
                 .setCompressionPoolThreadCount(8)
@@ -108,13 +106,6 @@ public void onMessages(DataReceivedEvent event) {
                 } else {
                     logger.info("Message received. SeqNo={}, offset={}", message.getSeqNo(), message.getOffset());
                 }
-                if (lastSeqNo > message.getSeqNo()) {
-                    logger.error("Received a message with seqNo {}. Previously got a message with seqNo {}",
-                            message.getSeqNo(), lastSeqNo);
-                    messageReceivedFuture.complete(null);
-                } else {
-                    lastSeqNo = message.getSeqNo();
-                }
                 message.commit().thenRun(() -> {
                     logger.info("Message committed");
                     if (messageCounter.incrementAndGet() >= MESSAGES_COUNT) {

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

@@ -7,7 +7,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.examples.SimpleExample;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.read.DecompressionException;
 import tech.ydb.topic.read.Message;
@@ -18,11 +17,11 @@
 /**
  * @author Nikolay Perfilov
  */
-public class ReadSync extends SimpleExample {
+public class ReadSync extends SimpleTopicExample {
     private static final Logger logger = LoggerFactory.getLogger(ReadSync.class);
 
     @Override
-    protected void run(GrpcTransport transport, String pathPrefix) {
+    protected void run(GrpcTransport transport) {
 
         try (TopicClient topicClient = TopicClient.newClient(transport).build()) {
 

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

@@ -19,7 +19,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.examples.SimpleExample;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.read.AsyncReader;
@@ -40,7 +39,7 @@
 /**
  * @author Nikolay Perfilov
  */
-public class ReadWriteWorkload extends SimpleExample {
+public class ReadWriteWorkload extends SimpleTopicExample {
     private static final Logger logger = LoggerFactory.getLogger(ReadWriteWorkload.class);
     private static final int WRITE_TIMEOUT_SECONDS = 60;
     private static final int MESSAGE_LENGTH_BYTES = 10_000_000; // 10 Mb
@@ -53,12 +52,11 @@ public class ReadWriteWorkload extends SimpleExample {
     private final AtomicInteger messagesReceived = new AtomicInteger(0);
     private final AtomicInteger messagesCommitted = new AtomicInteger(0);
     private final AtomicLong bytesWritten = new AtomicLong(0);
-    private long lastSeqNo = -1;
     CountDownLatch writeFinishedLatch = new CountDownLatch(1);
     CountDownLatch readFinishedLatch = new CountDownLatch(1);
 
     @Override
-    protected void run(GrpcTransport transport, String pathPrefix) {
+    protected void run(GrpcTransport transport) {
 
         ExecutorService compressionExecutor = Executors.newFixedThreadPool(10);
         AtomicBoolean timeToStopWriting = new AtomicBoolean(false);
@@ -154,9 +152,8 @@ protected void run(GrpcTransport transport, String pathPrefix) {
             };
 
             Runnable readingThread = () -> {
-                String consumerName = "consumer1";
                 ReaderSettings readerSettings = ReaderSettings.newBuilder()
-                        .setConsumerName(consumerName)
+                        .setConsumerName(CONSUMER_NAME)
                         .addTopic(TopicReadSettings.newBuilder()
                                 .setPath(TOPIC_NAME)
                                 .setReadFrom(Instant.now().minus(Duration.ofHours(24)))
@@ -274,12 +271,6 @@ public void onMessages(DataReceivedEvent event) {
                 } else {
                     logger.debug("Message received. SeqNo={}, offset={}", message.getSeqNo(), message.getOffset());
                 }
-                if (lastSeqNo > message.getSeqNo()) {
-                    logger.error("Received a message with seqNo {}. Previously got a message with seqNo {}",
-                            message.getSeqNo(), lastSeqNo);
-                } else {
-                    lastSeqNo = message.getSeqNo();
-                }
                 message.commit().thenRun(() -> {
                     logger.trace("Message committed");
                     unreadMessagesCount.decrementAndGet();

ydb-cookbook/src/main/java/tech/ydb/examples/topic/SimpleTopicExample.java

@@ -0,0 +1,40 @@
+package tech.ydb.examples.topic;
+
+import tech.ydb.auth.iam.CloudAuthHelper;
+import tech.ydb.core.grpc.GrpcTransport;
+
+
+/**
+ * @author Nikolay Perfilov
+ */
+public abstract class SimpleTopicExample {
+    protected static final String TOPIC_NAME = System.getenv("YDB_TOPIC_NAME");
+    protected static final String CONSUMER_NAME = System.getenv("YDB_CONSUMER_NAME");
+
+    protected void doMain(String[] args) {
+        if (args.length > 1) {
+            System.err.println("Too many arguments");
+            return;
+        }
+        String connString;
+        if (args.length == 1) {
+            connString = args[0];
+        } else {
+            System.err.println("Pass <connection-string> as an argument. " +
+                    "Example: some.host.name.com:2135?database=/Root\n");
+            return;
+        }
+
+        System.out.println("connection-string: " + connString + "\n");
+
+        try (GrpcTransport transport = GrpcTransport.forConnectionString(connString)
+                .withAuthProvider(CloudAuthHelper.getAuthProviderFromEnviron())
+                .build()) {
+            run(transport);
+        } catch (Throwable t) {
+            t.printStackTrace();
+        }
+    }
+
+    protected abstract void run(GrpcTransport transport);
+}

ydb-cookbook/src/main/java/tech/ydb/examples/topic/WriteAsync.java

@@ -10,7 +10,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.examples.SimpleExample;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.settings.WriterSettings;
@@ -22,13 +21,13 @@
 /**
  * @author Nikolay Perfilov
  */
-public class WriteAsync extends SimpleExample {
+public class WriteAsync extends SimpleTopicExample {
     private static final Logger logger = LoggerFactory.getLogger(WriteAsync.class);
     private static final int MESSAGES_COUNT = 5;
     private static final int WAIT_TIMEOUT_SECONDS = 60;
 
     @Override
-    protected void run(GrpcTransport transport, String pathPrefix) {
+    protected void run(GrpcTransport transport) {
         String producerId = "messageGroup1";
         String messageGroupId = "messageGroup1";
 

ydb-cookbook/src/main/java/tech/ydb/examples/topic/WriteSync.java

@@ -7,7 +7,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import tech.ydb.core.grpc.GrpcTransport;
-import tech.ydb.examples.SimpleExample;
 import tech.ydb.topic.TopicClient;
 import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.settings.WriterSettings;
@@ -17,16 +16,15 @@
 /**
  * @author Nikolay Perfilov
  */
-public class WriteSync extends SimpleExample {
+public class WriteSync extends SimpleTopicExample {
     private static final Logger logger = LoggerFactory.getLogger(WriteSync.class);
 
     @Override
-    protected void run(GrpcTransport transport, String pathPrefix) {
+    protected void run(GrpcTransport transport) {
         String producerId = "messageGroup1";
         String messageGroupId = "messageGroup1";
 
         try (TopicClient topicClient = TopicClient.newClient(transport).build()) {
-
             WriterSettings settings = WriterSettings.newBuilder()
                     .setTopicPath(TOPIC_NAME)
                     .setProducerId(producerId)