Merge pull request #215 from ydb-platform/release_v2.1.10

Release v2.1.10

CHANGELOG.md

@@ -1,3 +1,11 @@
+## 2.1.10 ##
+
+* Topics: Added message metadata support
+* Topics: Added support for reading without a consumer
+* Topics: Fixed a bug where onReaderClosed was not called on async reader shutdown
+* Topics: Small logging improvements
+* Added batch limit support for readTable
+
 ## 2.1.9 ##
 
 * Topics: Fixed a bug where first commit was not getting commitResponse if a user had sent a custom StartPartitionSessionResponse

README.md

@@ -30,7 +30,7 @@ Firstly you can import YDB Java BOM to specify correct versions of SDK modules.
         <dependency>
             <groupId>tech.ydb</groupId>
             <artifactId>ydb-sdk-bom</artifactId>
-            <version>2.1.9</version>
+            <version>2.1.10</version>
             <type>pom</type>
             <scope>import</scope>
         </dependency>

bom/pom.xml

@@ -6,7 +6,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>tech.ydb</groupId>
-    <version>2.1.9</version>
+    <version>2.1.10</version>
     <artifactId>ydb-sdk-bom</artifactId>
     <name>Java SDK Bill of Materials</name>
     <description>Java SDK Bill of Materials (BOM)</description>
@@ -15,7 +15,7 @@
 
     <properties>
         <ydb-auth-api.version>1.0.0</ydb-auth-api.version>
-        <ydb-proto-api.version>1.4.1</ydb-proto-api.version>
+        <ydb-proto-api.version>1.5.2</ydb-proto-api.version>
         <yc-auth.version>2.1.0</yc-auth.version>
     </properties>
 

coordination/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
     </parent>
 
     <artifactId>ydb-sdk-coordination</artifactId>

core/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
     </parent>
 
     <artifactId>ydb-sdk-core</artifactId>

core/src/main/java/tech/ydb/core/StatusCode.java

@@ -36,6 +36,7 @@ public enum StatusCode {
     UNDETERMINED(SERVER_STATUSES_FIRST + 170),
     UNSUPPORTED(SERVER_STATUSES_FIRST + 180),
     SESSION_BUSY(SERVER_STATUSES_FIRST + 190),
+    EXTERNAL_ERROR(SERVER_STATUSES_FIRST + 200),
 
     // Client statuses
     /** Cannot connect or unrecoverable network error. (map from gRPC UNAVAILABLE) */
@@ -129,6 +130,7 @@ public static StatusCode fromProto(StatusIds.StatusCode code) {
             case UNDETERMINED: return UNDETERMINED;
             case UNSUPPORTED: return UNSUPPORTED;
             case SESSION_BUSY: return SESSION_BUSY;
+            case EXTERNAL_ERROR: return EXTERNAL_ERROR;
             default:
                 return UNUSED_STATUS;
         }

core/src/main/resources/version.properties

@@ -1 +1 @@
-version=2.1.9
+version=2.1.10

pom.xml

@@ -8,7 +8,7 @@
 
     <groupId>tech.ydb</groupId>
     <artifactId>ydb-sdk-parent</artifactId>
-    <version>2.1.9</version>
+    <version>2.1.10</version>
 
     <name>Java SDK for YDB</name>
     <description>Java SDK for YDB</description>

scheme/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
     </parent>
 
     <artifactId>ydb-sdk-scheme</artifactId>

table/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
     </parent>
 
     <artifactId>ydb-sdk-table</artifactId>

table/src/main/java/tech/ydb/table/impl/BaseSession.java

@@ -741,7 +741,9 @@ public GrpcReadStream<ReadTablePart> executeReadTable(String tablePath, ReadTabl
                 .setSessionId(id)
                 .setPath(tablePath)
                 .setOrdered(settings.isOrdered())
-                .setRowLimit(settings.getRowLimit());
+                .setRowLimit(settings.getRowLimit())
+                .setBatchLimitBytes(settings.batchLimitBytes())
+                .setBatchLimitRows(settings.batchLimitRows());
 
         Value<?> fromKey = settings.getFromKey();
         if (fromKey != null) {

table/src/main/java/tech/ydb/table/settings/ReadTableSettings.java

@@ -29,6 +29,8 @@ public class ReadTableSettings extends BaseRequestSettings {
     private final boolean toInclusive;
     private final int rowLimit;
     private final ImmutableList<String> columns;
+    private final int batchLimitBytes;
+    private final int batchLimitRows;
 
     private ReadTableSettings(Builder builder) {
         super(builder);
@@ -39,6 +41,8 @@ private ReadTableSettings(Builder builder) {
         this.toInclusive = builder.toInclusive;
         this.rowLimit = builder.rowLimit;
         this.columns = ImmutableList.copyOf(builder.columns);
+        this.batchLimitBytes = builder.batchLimitBytes;
+        this.batchLimitRows = builder.batchLimitRows;
     }
 
     public static Builder newBuilder() {
@@ -75,6 +79,14 @@ public ImmutableList<String> getColumns() {
         return columns;
     }
 
+    public int batchLimitBytes() {
+        return batchLimitBytes;
+    }
+
+    public int batchLimitRows() {
+        return batchLimitRows;
+    }
+
     /**
      * BUILDER
      */
@@ -86,6 +98,8 @@ public static final class Builder extends BaseBuilder<Builder> {
         private boolean toInclusive = false;
         private int rowLimit = 0;
         private List<String> columns = Collections.emptyList();
+        private int batchLimitBytes = 0;
+        private int batchLimitRows = 0;
 
         public Builder orderedRead(boolean ordered) {
             this.ordered = ordered;
@@ -160,6 +174,16 @@ public Builder column(String column) {
             return this;
         }
 
+        public Builder batchLimitBytes(int batchLimitBytes) {
+            this.batchLimitBytes = batchLimitBytes;
+            return this;
+        }
+
+        public Builder batchLimitRows(int batchLimitRows) {
+            this.batchLimitRows = batchLimitRows;
+            return this;
+        }
+
         @Override
         public ReadTableSettings build() {
             return new ReadTableSettings(this);

tests/common/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
 

tests/junit4-support/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
 

tests/junit5-support/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
 

topic/pom.xml

@@ -8,7 +8,7 @@
     <parent>
         <groupId>tech.ydb</groupId>
         <artifactId>ydb-sdk-parent</artifactId>
-        <version>2.1.9</version>
+        <version>2.1.10</version>
     </parent>
 
     <artifactId>ydb-sdk-topic</artifactId>

topic/src/main/java/tech/ydb/topic/description/MetadataItem.java

@@ -0,0 +1,24 @@
+package tech.ydb.topic.description;
+
+import javax.annotation.Nonnull;
+
+/**
+ * @author Nikolay Perfilov
+ */
+public class MetadataItem {
+    private final String key;
+    private final byte[] value;
+
+    public MetadataItem(@Nonnull String key, byte[] value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    public String getKey() {
+        return key;
+    }
+
+    public byte[] getValue() {
+        return value;
+    }
+}

topic/src/main/java/tech/ydb/topic/read/Message.java

@@ -1,11 +1,14 @@
 package tech.ydb.topic.read;
 
 import java.time.Instant;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import io.grpc.ExperimentalApi;
 
+import tech.ydb.topic.description.MetadataItem;
+
 /**
  * @author Nikolay Perfilov
  */
@@ -52,6 +55,11 @@ public interface Message {
      */
     Instant getWrittenAt();
 
+    /**
+     * @return message metadata items
+     */
+    List<MetadataItem> getMetadataItems();
+
     /**
      * @return Partition session of this message
      */

topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java

@@ -119,21 +119,20 @@ protected void handleClosePartitionSession(tech.ydb.topic.read.PartitionSession
         });
     }
 
-    @Override
-    protected void handleCloseReader() {
+    protected void handleReaderClosed() {
         handlerExecutor.execute(() -> {
             eventHandler.onReaderClosed(new ReaderClosedEvent());
         });
     }
 
     @Override
-    protected CompletableFuture<Void> shutdownImpl() {
-        return super.shutdownImpl().whenComplete((res, th) -> {
-            if (defaultHandlerExecutorService != null) {
-                logger.debug("Shutting down default handler executor");
-                defaultHandlerExecutorService.shutdown();
-            }
-        });
+    protected void onShutdown(String reason) {
+        super.onShutdown(reason);
+        handleReaderClosed();
+        if (defaultHandlerExecutorService != null) {
+            logger.debug("Shutting down default handler executor");
+            defaultHandlerExecutorService.shutdown();
+        }
     }
 
     @Override

topic/src/main/java/tech/ydb/topic/read/impl/MessageImpl.java

@@ -2,9 +2,11 @@
 
 import java.io.IOException;
 import java.time.Instant;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
+import tech.ydb.topic.description.MetadataItem;
 import tech.ydb.topic.read.DecompressionException;
 import tech.ydb.topic.read.Message;
 import tech.ydb.topic.read.OffsetsRange;
@@ -22,6 +24,7 @@ public class MessageImpl implements Message {
     private final String messageGroupId;
     private final BatchMeta batchMeta;
     private final PartitionSessionImpl partitionSession;
+    private List<MetadataItem> metadataItems;
     private final OffsetsRange offsetsToCommit;
     private final CommitterImpl committer;
     private boolean isDecompressed = false;
@@ -36,6 +39,7 @@ private MessageImpl(Builder builder) {
         this.messageGroupId = builder.messageGroupId;
         this.batchMeta = builder.batchMeta;
         this.partitionSession = builder.partitionSession;
+        this.metadataItems = builder.metadataItems;
         this.offsetsToCommit = new OffsetsRangeImpl(commitOffsetFrom, offset + 1);
         this.committer = new CommitterImpl(partitionSession, 1, offsetsToCommit);
     }
@@ -105,6 +109,11 @@ public PartitionSessionImpl getPartitionSessionImpl() {
         return partitionSession;
     }
 
+    @Override
+    public List<MetadataItem> getMetadataItems() {
+        return metadataItems;
+    }
+
     public void setDecompressed(boolean decompressed) {
         isDecompressed = decompressed;
     }
@@ -130,6 +139,7 @@ public static class Builder {
         private String messageGroupId;
         private BatchMeta batchMeta;
         private PartitionSessionImpl partitionSession;
+        private List<MetadataItem> metadataItems;
 
         public Builder setData(byte[] data) {
             this.data = data;
@@ -171,6 +181,11 @@ public Builder setPartitionSession(PartitionSessionImpl partitionSession) {
             return this;
         }
 
+        public Builder setMetadataItems(List<MetadataItem> metadataItems) {
+            this.metadataItems = metadataItems;
+            return this;
+        }
+
         public MessageImpl build() {
             return new MessageImpl(this);
         }

topic/src/main/java/tech/ydb/topic/read/impl/PartitionSessionImpl.java

@@ -14,13 +14,15 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import tech.ydb.core.utils.ProtobufUtils;
 import tech.ydb.proto.topic.YdbTopic;
 import tech.ydb.topic.description.Codec;
+import tech.ydb.topic.description.MetadataItem;
 import tech.ydb.topic.read.Message;
 import tech.ydb.topic.read.OffsetsRange;
 import tech.ydb.topic.read.PartitionSession;
@@ -140,6 +142,11 @@ public CompletableFuture<Void> addBatches(List<YdbTopic.StreamReadMessage.ReadRe
                         .setCommitOffsetFrom(commitOffsetFrom)
                         .setCreatedAt(ProtobufUtils.protoToInstant(messageData.getCreatedAt()))
                         .setMessageGroupId(messageData.getMessageGroupId())
+                        .setMetadataItems(messageData.getMetadataItemsList()
+                                .stream()
+                                .map(metadataItem -> new MetadataItem(metadataItem.getKey(),
+                                        metadataItem.getValue().toByteArray()))
+                                .collect(Collectors.toList()))
                         .build()
                 );
             });
@@ -245,10 +252,12 @@ public void handleCommitResponse(long committedOffset) {
             return;
         }
         Map<Long, CompletableFuture<Void>> futuresToComplete = commitFutures.headMap(committedOffset, true);
-        logger.info("[{}] Commit response received for partition session {} (partition {}). Committed offset: {}" +
-                        ". Previous committed offset: {} (diff is {} message(s)). Completing {} commit futures",
-                path, id, partitionId, committedOffset, lastCommittedOffset, committedOffset - lastCommittedOffset,
-                futuresToComplete.size());
+        if (logger.isDebugEnabled()) {
+            logger.debug("[{}] Commit response received for partition session {} (partition {}). Committed offset: {}" +
+                            ". Previous committed offset: {} (diff is {} message(s)). Completing {} commit futures",
+                    path, id, partitionId, committedOffset, lastCommittedOffset, committedOffset - lastCommittedOffset,
+                    futuresToComplete.size());
+        }
         lastCommittedOffset = committedOffset;
         futuresToComplete.values().forEach(future -> future.complete(null));
         futuresToComplete.clear();