Skip to content

Commit a71c357

Browse files
KirillKurdyukovKirillKurdyukov
committed
Merge branch 'master' into version
# Conflicts: # jdbc/spring-data-jpa/pom.xml
2 parents e36ebc9 + c49e92b commit a71c357

File tree

11 files changed

+208
-143
lines changed

11 files changed

+208
-143
lines changed

‎jdbc/spring-data-jpa/pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
<properties>
1313
<maven.compiler.release>17</maven.compiler.release>
1414
<kotlin.version>1.9.22</kotlin.version>
15-
<hibernate.ydb.dialect.version>1.1.0</hibernate.ydb.dialect.version>
15+
<hibernate.ydb.dialect.version>1.4.0</hibernate.ydb.dialect.version>
1616
<spring.boot.version>3.2.1</spring.boot.version>
1717
</properties>
1818
<dependencies>

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

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,27 @@
11
package tech.ydb.examples;
22

3+
import java.time.Duration;
4+
35
import tech.ydb.auth.iam.CloudAuthHelper;
6+
import tech.ydb.core.Result;
7+
import tech.ydb.core.Status;
48
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;
516

617

718
/**
819
* @author Sergey Polovko
920
* @author Nikolay Perfilov
1021
*/
1122
public abstract class SimpleExample {
12-
protected static final String TOPIC_NAME = "test-topic";
13-
protected static final String CONSUMER_NAME = "test-consumer";
23+
protected static final String TOPIC_NAME = System.getenv("YDB_TOPIC_NAME");
24+
protected static final String CONSUMER_NAME = System.getenv("YDB_CONSUMER_NAME");
1425

1526
protected void doMain(String[] args) {
1627
if (args.length > 1) {

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,9 @@ public void onPartitionSessionClosed(PartitionSessionClosedEvent event) {
160160
@Override
161161
public void onReaderClosed(ReaderClosedEvent event) {
162162
logger.info("Reader is closed.");
163+
if (!messageReceivedFuture.isDone()) {
164+
messageReceivedFuture.complete(null);
165+
}
163166
}
164167
}
165168

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

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,14 +3,12 @@
33
import java.nio.charset.StandardCharsets;
44
import java.time.Duration;
55
import java.time.Instant;
6-
import java.util.List;
76

87
import org.slf4j.Logger;
98
import org.slf4j.LoggerFactory;
109
import tech.ydb.core.grpc.GrpcTransport;
1110
import tech.ydb.examples.SimpleExample;
1211
import tech.ydb.topic.TopicClient;
13-
import tech.ydb.topic.description.MetadataItem;
1412
import tech.ydb.topic.read.DecompressionException;
1513
import tech.ydb.topic.read.Message;
1614
import tech.ydb.topic.read.SyncReader;

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package tech.ydb.examples.topic;
22

3+
import java.nio.charset.StandardCharsets;
34
import java.text.DecimalFormat;
45
import java.time.Duration;
56
import java.time.Instant;
@@ -109,6 +110,9 @@ protected void run(GrpcTransport transport, String pathPrefix) {
109110
}
110111
logger.info("Received a signal to stop writing");
111112

113+
// Wait for all writes to receive a WriteAck before shutting down writer
114+
writer.flush();
115+
112116
try {
113117
writer.shutdown(SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS);
114118
} catch (TimeoutException exception) {
@@ -240,7 +244,7 @@ private class Handler extends AbstractReadEventHandler {
240244
public void onMessages(DataReceivedEvent event) {
241245
for (tech.ydb.topic.read.Message message : event.getMessages()) {
242246
messagesReceived.incrementAndGet();
243-
if (logger.isTraceEnabled()) {
247+
if (logger.isDebugEnabled()) {
244248
StringBuilder str = new StringBuilder("Message received");
245249
str.append("\n");
246250
str.append(" offset: ").append(message.getOffset()).append("\n")
@@ -251,13 +255,22 @@ public void onMessages(DataReceivedEvent event) {
251255
.append(" writtenAt: ").append(message.getWrittenAt()).append("\n")
252256
.append(" partitionSession: ").append(message.getPartitionSession().getId()).append("\n")
253257
.append(" partitionId: ").append(message.getPartitionSession().getPartitionId())
258+
.append("\n")
259+
.append(" metadataItems: ")
254260
.append("\n");
261+
message.getMetadataItems().forEach(item -> str
262+
.append(" key: \"")
263+
.append(item.getKey())
264+
.append("\", value: \"")
265+
.append(new String(item.getValue(), StandardCharsets.UTF_8))
266+
.append("\"")
267+
.append("\n"));
255268
if (!message.getWriteSessionMeta().isEmpty()) {
256269
str.append(" writeSessionMeta:\n");
257270
message.getWriteSessionMeta().forEach((key, value) ->
258271
str.append(" ").append(key).append(": ").append(value).append("\n"));
259272
}
260-
logger.trace(str.toString());
273+
logger.debug(str.toString());
261274
} else {
262275
logger.debug("Message received. SeqNo={}, offset={}", message.getSeqNo(), message.getOffset());
263276
}

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

Lines changed: 31 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package tech.ydb.examples.topic;
22

3+
import java.util.concurrent.CountDownLatch;
34
import java.util.concurrent.ExecutionException;
45
import java.util.concurrent.ExecutorService;
56
import java.util.concurrent.Executors;
@@ -23,6 +24,8 @@
2324
*/
2425
public class WriteAsync extends SimpleExample {
2526
private static final Logger logger = LoggerFactory.getLogger(WriteAsync.class);
27+
private static final int MESSAGES_COUNT = 5;
28+
private static final int WAIT_TIMEOUT_SECONDS = 60;
2629

2730
@Override
2831
protected void run(GrpcTransport transport, String pathPrefix) {
@@ -52,14 +55,17 @@ protected void run(GrpcTransport transport, String pathPrefix) {
5255
return null;
5356
});
5457

55-
for (int i = 1; i <= 5; i++) {
58+
// A latch to wait for all writes to receive a WriteAck before shutting down writer
59+
CountDownLatch writesInProgress = new CountDownLatch(MESSAGES_COUNT);
60+
61+
for (int i = 1; i <= MESSAGES_COUNT; i++) {
5662
final int index = i;
5763
try {
5864
String messageString = "message" + i;
5965
// Blocks until the message is put into sending buffer
6066
writer.send(Message.of(messageString.getBytes())).whenComplete((result, ex) -> {
6167
if (ex != null) {
62-
logger.error("Exception while sending message {}: ", index, ex);
68+
logger.error("Exception while sending a message {}: ", index, ex);
6369
} else {
6470
logger.info("Message {} ack received", index);
6571

@@ -76,20 +82,38 @@ protected void run(GrpcTransport transport, String pathPrefix) {
7682
break;
7783
}
7884
}
85+
writesInProgress.countDown();
7986
});
8087
} catch (QueueOverflowException exception) {
81-
logger.error("Queue overflow exception while sending message{}: ", index, exception);
82-
// Send queue is full. Need retry with backoff or skip
88+
logger.error("Queue overflow exception while sending a message{}: ", index, exception);
89+
// Send queue is full. Need to retry with backoff or skip
90+
writesInProgress.countDown();
8391
}
8492

8593
logger.info("Message {} is sent", index);
8694
}
8795

88-
long timeoutSeconds = 10;
8996
try {
90-
writer.shutdown().get(timeoutSeconds, TimeUnit.SECONDS);
97+
while (!writesInProgress.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
98+
logger.error("Writes are not finished in {} seconds", WAIT_TIMEOUT_SECONDS);
99+
}
100+
} catch (InterruptedException exception) {
101+
logger.error("Waiting for writes to finish was interrupted: ", exception);
102+
}
103+
104+
try {
105+
if (!writesInProgress.await(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
106+
logger.error("Writes are not finished in {} seconds", WAIT_TIMEOUT_SECONDS);
107+
}
108+
} catch (InterruptedException exception) {
109+
logger.error("Waiting for writes to finish was interrupted: ", exception);
110+
}
111+
112+
try {
113+
writer.shutdown().get(WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS);
91114
} catch (TimeoutException exception) {
92-
logger.error("Timeout exception during writer termination ({} seconds): ", timeoutSeconds, exception);
115+
logger.error("Timeout exception during writer termination ({} seconds): ", WAIT_TIMEOUT_SECONDS,
116+
exception);
93117
} catch (ExecutionException exception) {
94118
logger.error("Execution exception during writer termination: ", exception);
95119
} catch (InterruptedException exception) {

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,8 +67,10 @@ protected void run(GrpcTransport transport, String pathPrefix) {
6767
}
6868
}
6969

70+
// Wait for all writes to receive a WriteAck before shutting down writer
7071
writer.flush();
7172
logger.info("Flush finished");
73+
7274
long shutdownTimeoutSeconds = 10;
7375
try {
7476
writer.shutdown(shutdownTimeoutSeconds, TimeUnit.SECONDS);

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

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,8 +18,6 @@
1818
import tech.ydb.table.impl.PooledTableClient;
1919
import tech.ydb.table.rpc.grpc.GrpcTableRpc;
2020
import tech.ydb.table.transaction.TableTransaction;
21-
import tech.ydb.table.transaction.Transaction;
22-
import tech.ydb.table.transaction.TxControl;
2321
import tech.ydb.topic.TopicClient;
2422
import tech.ydb.topic.read.AsyncReader;
2523
import tech.ydb.topic.read.DecompressionException;
@@ -42,7 +40,7 @@
4240
public class TransactionReadAsync extends SimpleExample {
4341
private static final Logger logger = LoggerFactory.getLogger(TransactionReadAsync.class);
4442
private static final long MAX_MEMORY_USAGE_BYTES = 500 * 1024 * 1024; // 500 Mb
45-
private static final int MESSAGES_COUNT = 5;
43+
private static final int MESSAGES_COUNT = 1;
4644

4745
private final CompletableFuture<Void> messageReceivedFuture = new CompletableFuture<>();
4846
private TableClient tableClient;
@@ -135,11 +133,13 @@ public void onMessages(DataReceivedEvent event) {
135133
// creating session and transaction
136134
Result<Session> sessionResult = tableClient.createSession(Duration.ofSeconds(10)).join();
137135
if (!sessionResult.isSuccess()) {
138-
logger.error("Couldn't get session from pool: {}", sessionResult);
136+
logger.error("Couldn't get a session from the pool: {}", sessionResult);
139137
return; // retry or shutdown
140138
}
141139
Session session = sessionResult.getValue();
142-
TableTransaction transaction = session.createNewTransaction(TxMode.SERIALIZABLE_RW);
140+
TableTransaction transaction = session.beginTransaction(TxMode.SERIALIZABLE_RW)
141+
.join()
142+
.getValue();
143143

144144
// do something else in transaction
145145
transaction.executeDataQuery("SELECT 1").join();
@@ -199,6 +199,7 @@ public void onPartitionSessionClosed(PartitionSessionClosedEvent event) {
199199
@Override
200200
public void onReaderClosed(ReaderClosedEvent event) {
201201
logger.info("Reader is closed.");
202+
messageReceivedFuture.complete(null);
202203
}
203204
}
204205

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

Lines changed: 25 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -13,8 +13,6 @@
1313
import tech.ydb.table.Session;
1414
import tech.ydb.table.TableClient;
1515
import tech.ydb.table.transaction.TableTransaction;
16-
import tech.ydb.table.transaction.Transaction;
17-
import tech.ydb.table.transaction.TxControl;
1816
import tech.ydb.topic.TopicClient;
1917
import tech.ydb.topic.read.DecompressionException;
2018
import tech.ydb.topic.read.Message;
@@ -49,37 +47,34 @@ protected void run(GrpcTransport transport, String pathPrefix) {
4947
reader.init();
5048

5149
try {
52-
// Reading 5 messages
53-
for (int i = 0; i < 5; i++) {
54-
// creating session and transaction
55-
Result<Session> sessionResult = tableClient.createSession(Duration.ofSeconds(10)).join();
56-
if (!sessionResult.isSuccess()) {
57-
logger.error("Couldn't get session from pool: {}", sessionResult);
58-
return; // retry or shutdown
59-
}
60-
Session session = sessionResult.getValue();
61-
TableTransaction transaction = session.createNewTransaction(TxMode.SERIALIZABLE_RW);
62-
63-
// do something else in transaction
64-
transaction.executeDataQuery("SELECT 1").join();
65-
// analyzeQueryResultIfNeeded();
50+
// creating session and transaction
51+
Result<Session> sessionResult = tableClient.createSession(Duration.ofSeconds(10)).join();
52+
if (!sessionResult.isSuccess()) {
53+
logger.error("Couldn't a get session from the pool: {}", sessionResult);
54+
return; // retry or shutdown
55+
}
56+
Session session = sessionResult.getValue();
57+
TableTransaction transaction = session.createNewTransaction(TxMode.SERIALIZABLE_RW);
6658

67-
//Session session
68-
Message message = reader.receive(ReceiveSettings.newBuilder()
69-
.setTransaction(transaction)
70-
.build());
71-
byte[] messageData;
72-
try {
73-
messageData = message.getData();
74-
} catch (DecompressionException e) {
75-
logger.warn("Decompression exception while receiving a message: ", e);
76-
messageData = e.getRawData();
77-
}
78-
logger.info("Message received: {}", new String(messageData, StandardCharsets.UTF_8));
59+
// do something else in transaction
60+
transaction.executeDataQuery("SELECT 1").join();
61+
// analyzeQueryResultIfNeeded();
7962

80-
transaction.commit().join();
81-
// analyze commit status
63+
//Session session
64+
Message message = reader.receive(ReceiveSettings.newBuilder()
65+
.setTransaction(transaction)
66+
.build());
67+
byte[] messageData;
68+
try {
69+
messageData = message.getData();
70+
} catch (DecompressionException e) {
71+
logger.warn("Decompression exception while receiving a message: ", e);
72+
messageData = e.getRawData();
8273
}
74+
logger.info("Message received: {}", new String(messageData, StandardCharsets.UTF_8));
75+
76+
transaction.commit().join();
77+
// analyze commit status
8378
} catch (InterruptedException exception) {
8479
logger.error("Interrupted exception while waiting for message: ", exception);
8580
}

0 commit comments

Comments
 (0)