From e72c8dc7df49c98dbb5b0c1e700011edf1eee769 Mon Sep 17 00:00:00 2001 From: Alexandr Gorshenin Date: Fri, 4 Sep 2026 09:01:53 +0100 Subject: [PATCH 1/2] Added cdc timestamp test --- .../tech/ydb/topic/ChangefeedTopicTest.java | 83 +++++++++++++++++++ 1 file changed, 83 insertions(+) diff --git a/topic/src/test/java/tech/ydb/topic/ChangefeedTopicTest.java b/topic/src/test/java/tech/ydb/topic/ChangefeedTopicTest.java index f18daf93a..518d933b1 100644 --- a/topic/src/test/java/tech/ydb/topic/ChangefeedTopicTest.java +++ b/topic/src/test/java/tech/ydb/topic/ChangefeedTopicTest.java @@ -1,5 +1,6 @@ package tech.ydb.topic; +import java.time.Duration; import java.util.Iterator; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; @@ -13,14 +14,22 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.Timeout; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import tech.ydb.common.transaction.TxMode; import tech.ydb.core.Status; +import tech.ydb.table.Session; import tech.ydb.table.SessionRetryContext; import tech.ydb.table.TableClient; import tech.ydb.table.query.Params; +import tech.ydb.table.result.ResultSetReader; +import tech.ydb.table.settings.ReadTableSettings; +import tech.ydb.table.transaction.TableTransaction; import tech.ydb.table.transaction.TxControl; import tech.ydb.test.junit4.GrpcTransportRule; import tech.ydb.topic.read.AsyncReader; +import tech.ydb.topic.read.Message; import tech.ydb.topic.read.events.DataReceivedEvent; import tech.ydb.topic.read.events.PartitionSessionClosedEvent; import tech.ydb.topic.read.events.ReadEventHandler; @@ -37,6 +46,7 @@ * @author Aleksandr Gorshenin */ public class ChangefeedTopicTest { + private final static Logger logger = LoggerFactory.getLogger(ChangefeedTopicTest.class); private final static String TEST_TABLE = "changefeed_table"; private final static String TEST_CHANGEFEED = "updates"; private final static String TEST_CONSUMER = "consumer"; @@ -139,6 +149,79 @@ public void changefeedReadTest() throws Exception { handler2.assertEvents("INIT", "START", "DATA1", "CLOSED", "READER CLOSED"); } + @Test + public void changefeedTimestampTest() throws Exception { + SessionRetryContext retryCtx = SessionRetryContext.create(tableClient).build(); + String tablePath = ydbTransport.getDatabase() + "/" + "timestamp_test"; + String changefeedPath = tablePath + "/" + TEST_CHANGEFEED; + + retryCtx.supplyStatus(s -> s.executeSchemeQuery( + "DROP TABLE IF EXISTS `" + tablePath + "`;" + )).join().expectSuccess("cannot drop table"); + + retryCtx.supplyStatus(s -> s.executeSchemeQuery( + "CREATE TABLE `" + tablePath + "` (id Uint32, value Text, PRIMARY KEY (id));" + )).join().expectSuccess("cannot create table"); + + retryCtx.supplyResult(s -> s.executeDataQuery( + "INSERT INTO `" + tablePath + "` (id, value) VALUES (1, '1'), (2, '2');", + TxControl.serializableRw(), Params.empty() + )).join().getStatus().expectSuccess("cannot insert data"); + + retryCtx.supplyStatus(s -> s.executeSchemeQuery( + "ALTER TABLE `" + tablePath + "` " + "ADD CHANGEFEED " + TEST_CHANGEFEED + + " WITH (FORMAT = 'JSON', MODE = 'NEW_IMAGE', VIRTUAL_TIMESTAMPS = true);" + )).join().expectSuccess("cannot alter table"); + + retryCtx.supplyStatus(s -> s.executeSchemeQuery("" + + "ALTER TOPIC `" + changefeedPath + "` ADD CONSUMER " + TEST_CONSUMER + )).join().expectSuccess("cannot alter changefeed"); + + Session txs = tableClient.createSession(Duration.ofSeconds(5)).join().getValue(); + TableTransaction tx = txs.createNewTransaction(TxMode.SERIALIZABLE_RW); + tx.executeDataQuery("INSERT INTO `" + tablePath + "` (id, value) VALUES (3, '3'), (4, '4');").join() + .getStatus().expectSuccess(); + + ReaderSettings rs = ReaderSettings.newBuilder() + .addTopic(TopicReadSettings.newBuilder().setPath(changefeedPath).build()) + .setConsumerName(TEST_CONSUMER) + .build(); + + CountDownLatch latch = new CountDownLatch(4); + AsyncReader reader1 = topicClient.createAsyncReader(rs, ReadEventHandlersSettings.newBuilder() + .setEventHandler(new ReadEventHandler() { + @Override + public void onSessionStarted(SessionStartedEvent event) { + try (Session s = tableClient.createSession(Duration.ofSeconds(5)).join().getValue()) { + s.executeReadTable(tablePath, ReadTableSettings.newBuilder().build()).start(part -> { + ResultSetReader rsr = part.getResultSetReader(); + while (rsr.next()) { + logger.info("read row {} -> [{}, {}]", + rsr.getColumn("id").getUint32(), + part.getVirtualTimestamp().getPlanStep(), + part.getVirtualTimestamp().getTxId() + ); + latch.countDown(); + } + }).join().expectSuccess(); + } + tx.commit().join().expectSuccess(); + } + + @Override + public void onMessages(DataReceivedEvent event) { + for (Message msg : event.getMessages()) { + logger.info("cdc read {}", new String(msg.getData())); + latch.countDown(); + } + } + }).build()); + + reader1.init().join(); + Assert.assertTrue(latch.await(10, TimeUnit.SECONDS)); + reader1.shutdown().join(); + } + private static class TestHandler implements ReadEventHandler { private final CountDownLatch dataLatch; private final CountDownLatch closedLatch; From cbf9434fc9940df81a0b5dbc4edcde46539219f2 Mon Sep 17 00:00:00 2001 From: Alexandr Gorshenin Date: Fri, 4 Sep 2026 11:57:30 +0100 Subject: [PATCH 2/2] Added various tx writer tests --- .../topic/TopicWritersIntegrationTest.java | 209 ++++++++++++++++++ 1 file changed, 209 insertions(+) diff --git a/topic/src/test/java/tech/ydb/topic/TopicWritersIntegrationTest.java b/topic/src/test/java/tech/ydb/topic/TopicWritersIntegrationTest.java index b739cfdc9..6f9cb6edf 100644 --- a/topic/src/test/java/tech/ydb/topic/TopicWritersIntegrationTest.java +++ b/topic/src/test/java/tech/ydb/topic/TopicWritersIntegrationTest.java @@ -1,5 +1,6 @@ package tech.ydb.topic; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -21,9 +22,13 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import tech.ydb.common.transaction.TxMode; import tech.ydb.core.StatusCode; import tech.ydb.core.UnexpectedResultException; import tech.ydb.core.utils.FutureTools; +import tech.ydb.table.Session; +import tech.ydb.table.TableClient; +import tech.ydb.table.transaction.TableTransaction; import tech.ydb.test.junit4.GrpcTransportRule; import tech.ydb.topic.description.Consumer; import tech.ydb.topic.read.DeferredCommitter; @@ -31,6 +36,7 @@ import tech.ydb.topic.settings.CreateTopicSettings; import tech.ydb.topic.settings.PartitioningSettings; import tech.ydb.topic.settings.ReaderSettings; +import tech.ydb.topic.settings.SendSettings; import tech.ydb.topic.settings.TopicReadSettings; import tech.ydb.topic.settings.TopicRetryConfig; import tech.ydb.topic.settings.WriterSettings; @@ -413,4 +419,207 @@ public void wrongDirectWriteTest() throws Exception { ex2.getCause().getMessage() ); } + + @Test + public void txWriteTest() throws Exception { + + WriterSettings settings = WriterSettings.newBuilder() + .setTopicPath(TEST_TOPIC) + .setProducerId(TEST_PRODUCER1) + .setRetryConfig(TopicRetryConfig.STANDARD) + .build(); + + byte[] msg1 = new byte[1000]; + byte[] msg2 = new byte[1001]; + byte[] msg3 = new byte[1002]; + byte[] msg4 = new byte[1003]; + Arrays.fill(msg1, (byte) 0x10); + Arrays.fill(msg2, (byte) 0x11); + Arrays.fill(msg3, (byte) 0x50); + Arrays.fill(msg4, (byte) 0x60); + + SyncWriter writer = client.createSyncWriter(settings); + writer.initAndWait(); + + try ( + TableClient table = TableClient.newClient(ydbTransport).build(); + Session s1 = table.createSession(Duration.ofSeconds(5)).join().getValue(); + Session s2 = table.createSession(Duration.ofSeconds(5)).join().getValue()) { + + TableTransaction tx1 = s1.beginTransaction(TxMode.SERIALIZABLE_RW).join().getValue(); + TableTransaction tx2 = s2.beginTransaction(TxMode.SERIALIZABLE_RW).join().getValue(); + + writer.send( + Message.newBuilder().setData(msg1).setSeqNo(1).build(), + SendSettings.newBuilder().setTransaction(tx1).build() + ); + writer.send( + Message.newBuilder().setData(msg2).setSeqNo(2).build(), + SendSettings.newBuilder().setTransaction(tx2).build() + ); + writer.send( + Message.newBuilder().setData(msg3).setSeqNo(3).build(), + SendSettings.newBuilder().setTransaction(tx1).build() + ); + writer.send( + Message.newBuilder().setData(msg4).setSeqNo(4).build(), + SendSettings.newBuilder().setTransaction(tx2).build() + ); + + writer.flush(); + + tx2.commit().join().expectSuccess(); + Assert.assertEquals(StatusCode.ABORTED, tx1.commit().join().getCode()); + + writer.send( + Message.newBuilder().setData(msg1).setSeqNo(5).build() + ); + writer.send( + Message.newBuilder().setData(msg2).setSeqNo(6).build() + ); + + writer.flush(); + + TableTransaction tx3 = s2.beginTransaction(TxMode.SERIALIZABLE_RW).join().getValue(); + + writer.send( + Message.newBuilder().setData(msg3).setSeqNo(7).build(), + SendSettings.newBuilder().setTransaction(tx3).build() + ); + writer.send( + Message.newBuilder().setData(msg4).setSeqNo(8).build(), + SendSettings.newBuilder().setTransaction(tx3).build() + ); + + writer.flush(); + + tx3.commit().join().expectSuccess(); + writer.shutdown(1, TimeUnit.SECONDS); + } + + assertTopicContent(Arrays.asList(msg2, msg4, msg1, msg2, msg3, msg4)); + } + + @Test + public void invalidTxWriteTest() throws Exception { + List realErrors = new ArrayList<>(); + + WriterSettings settings = WriterSettings.newBuilder() + .setTopicPath(TEST_TOPIC) + .setProducerId(TEST_PRODUCER1) + .setRetryConfig(TopicRetryConfig.STANDARD) + .setErrorsHandler((st, th) -> { + if (st != null) { + realErrors.add(st.getCode()); + } + if (th != null) { + realErrors.add(StatusCode.CLIENT_INTERNAL_ERROR); + } + }) + .build(); + + byte[] msg1 = new byte[1000]; + byte[] msg2 = new byte[1001]; + byte[] msg3 = new byte[1002]; + byte[] msg4 = new byte[1003]; + Arrays.fill(msg1, (byte) 0x10); + Arrays.fill(msg2, (byte) 0x11); + Arrays.fill(msg3, (byte) 0x50); + Arrays.fill(msg4, (byte) 0x60); + + SyncWriter writer = client.createSyncWriter(settings); + writer.initAndWait(); + + try ( + TableClient table = TableClient.newClient(ydbTransport).build(); + Session s1 = table.createSession(Duration.ofSeconds(5)).join().getValue(); + Session s2 = table.createSession(Duration.ofSeconds(5)).join().getValue()) { + + TableTransaction tx1 = s1.beginTransaction(TxMode.SERIALIZABLE_RW).join().getValue(); + TableTransaction tx2 = s2.beginTransaction(TxMode.SERIALIZABLE_RW).join().getValue(); + + writer.send( + Message.newBuilder().setData(msg1).setSeqNo(1).build(), + SendSettings.newBuilder().setTransaction(tx1).build() + ); + writer.send( + Message.newBuilder().setData(msg2).setSeqNo(2).build(), + SendSettings.newBuilder().setTransaction(tx1).build() + ); + + tx2.rollback().join(); + + writer.send( + Message.newBuilder().setData(msg3).setSeqNo(3).build(), + SendSettings.newBuilder().setTransaction(tx2).build() + ); + writer.send( + Message.newBuilder().setData(msg4).setSeqNo(4).build(), + SendSettings.newBuilder().setTransaction(tx2).build() + ); + + writer.flush(); + writer.shutdown(1, TimeUnit.SECONDS); + + tx1.commit().join().expectSuccess(); + Assert.assertEquals(1, realErrors.size()); + Assert.assertEquals(StatusCode.NOT_FOUND, realErrors.get(0)); + } + + assertTopicContent(Arrays.asList(msg1, msg2)); + } + + @Test + public void txRetryWriteTest() throws Exception { + List realErrors = new ArrayList<>(); + PROXY.unavailableOnAckWithSeqNo(2); + + WriterSettings settings = WriterSettings.newBuilder() + .setTopicPath(TEST_TOPIC) + .setProducerId(TEST_PRODUCER1) + .setRetryConfig(TopicRetryConfig.STANDARD) + .setErrorsHandler((st, th) -> { + if (st != null) { + realErrors.add(st.getCode()); + } + if (th != null) { + realErrors.add(StatusCode.CLIENT_INTERNAL_ERROR); + } + }) + .build(); + + byte[] msg1 = new byte[1000]; + byte[] msg2 = new byte[1001]; + Arrays.fill(msg1, (byte) 0x10); + Arrays.fill(msg2, (byte) 0x11); + + AsyncWriter writer = client.createAsyncWriter(settings); + writer.init().join(); + + try ( + TableClient table = TableClient.newClient(ydbTransport).build(); + Session s1 = table.createSession(Duration.ofSeconds(5)).join().getValue()) { + + TableTransaction tx1 = s1.beginTransaction(TxMode.SERIALIZABLE_RW).join().getValue(); + + writer.send( + Message.newBuilder().setData(msg1).setSeqNo(1).build(), + SendSettings.newBuilder().setTransaction(tx1).build() + ); + CompletableFuture ack2 = writer.send( + Message.newBuilder().setData(msg2).setSeqNo(2).build(), + SendSettings.newBuilder().setTransaction(tx1).build() + ); + + Assert.assertEquals(WriteAck.State.ALREADY_WRITTEN, ack2.join().getState()); + + tx1.commit().join().expectSuccess(); + Assert.assertEquals(1, realErrors.size()); + Assert.assertEquals(StatusCode.TRANSPORT_UNAVAILABLE, realErrors.get(0)); + + writer.shutdown().join(); + } + + assertTopicContent(Arrays.asList(msg1, msg2)); + } }