Merge pull request #482 from alex268/update_commit_offsets2

Added support of readSessionId

Author: alex268

bom/pom.xml

@@ -15,7 +15,7 @@
 
     <properties>
         <ydb-auth-api.version>1.0.0</ydb-auth-api.version>
-        <ydb-proto-api.version>1.7.0</ydb-proto-api.version>
+        <ydb-proto-api.version>1.7.1</ydb-proto-api.version>
         <yc-auth.version>2.2.0</yc-auth.version>
     </properties>
 

topic/src/main/java/tech/ydb/topic/impl/TopicClientImpl.java

@@ -373,15 +373,19 @@ public AsyncReader createAsyncReader(ReaderSettings settings, ReadEventHandlersS
 
     @Override
     public CompletableFuture<Status> commitOffset(String path, CommitOffsetSettings settings) {
-        YdbTopic.CommitOffsetRequest request = YdbTopic.CommitOffsetRequest.newBuilder()
+        YdbTopic.CommitOffsetRequest.Builder request = YdbTopic.CommitOffsetRequest.newBuilder()
                 .setOperationParams(Operation.buildParams(settings))
                 .setOffset(settings.getOffset())
-                .setPath(path)
                 .setConsumer(settings.getConsumer())
                 .setPartitionId(settings.getPartitionId())
-                .build();
+                .setPath(path);
+
+        if (settings.getReadSessionId() != null) {
+            request.setReadSessionId(settings.getReadSessionId());
+        }
+
         final GrpcRequestSettings grpcRequestSettings = makeGrpcRequestSettings(settings);
-        return topicRpc.commitOffset(request, grpcRequestSettings);
+        return topicRpc.commitOffset(request.build(), grpcRequestSettings);
     }
 
     @Override

topic/src/main/java/tech/ydb/topic/settings/CommitOffsetSettings.java

@@ -9,12 +9,14 @@ public class CommitOffsetSettings extends OperationSettings {
     private final long partitionId;
     private final String consumer;
     private final long offset;
+    private final String readSessionId;
 
     private CommitOffsetSettings(Builder builder) {
         super(builder);
         this.partitionId = builder.partitionId;
         this.consumer = builder.consumer;
         this.offset = builder.offset;
+        this.readSessionId = builder.readSessionId;
     }
 
     public static Builder newBuilder() {
@@ -33,13 +35,18 @@ public long getOffset() {
         return offset;
     }
 
-    /**
+    public String getReadSessionId() {
+        return readSessionId;
+    }
+
+    /*
      * BUILDER
      */
     public static class Builder extends OperationBuilder<Builder> {
         private long partitionId = -1;
         private String consumer = null;
         private long offset = 0;
+        private String readSessionId = null;
 
         public Builder setPartitionId(long partitionId) {
             this.partitionId = partitionId;
@@ -56,6 +63,11 @@ public Builder setOffset(long offset) {
             return this;
         }
 
+        public Builder setReadSessionId(String sessionId) {
+            this.readSessionId = sessionId;
+            return this;
+        }
+
         @Override
         public CommitOffsetSettings build() {
             return new CommitOffsetSettings(this);