Skip to content

Added extra TxWriter and changeset reading tests #717

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
alex268 merged 2 commits into from
Sep 4, 2026
Merged

Added extra TxWriter and changeset reading tests #717

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e72c8dc
Added cdc timestamp test
alex268 Sep 4, 2026
cbf9434
Added various tx writer tests
alex268 Sep 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 83 additions & 0 deletions topic/src/test/java/tech/ydb/topic/ChangefeedTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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;
Expand All @@ -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";
Expand Down Expand Up @@ -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;
Expand Down
209 changes: 209 additions & 0 deletions topic/src/test/java/tech/ydb/topic/TopicWritersIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
package tech.ydb.topic;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
Expand All @@ -21,16 +22,21 @@
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;
import tech.ydb.topic.read.SyncReader;
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;
Expand Down Expand Up @@ -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<StatusCode> 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<StatusCode> 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<WriteAck> 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));
}
}
Loading