Support consumer version in ConsumerBuilder

.gitignore

@@ -3,3 +3,4 @@
 .ijwb
 bazel-*
 MODULE.bazel.lock
+.bazelbsp

src/consumer/Consumer.java

@@ -42,7 +42,7 @@
 public class Consumer extends AbstractService {
 
   /** Version of the client library. Should be kept in sync with Go client version. */
-  public static final String CLIENT_VERSION = "1.2.4";
+  public static final String CLIENT_VERSION = "1.2.5";
 
   private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
 

src/consumer/ConsumerBuilder.java

@@ -41,6 +41,7 @@ public class ConsumerBuilder {
   private @Nullable Integer capacityLimit;
   private @Nullable DomainKeyName[] keyNames;
   private @Nullable BooleanSupplier verbose;
+  private @Nullable String version;
 
   /**
    * Creates a new builder for a {@link Consumer} with default values. It will use the local IP
@@ -134,6 +135,16 @@ public ConsumerBuilder withVerboseLogging(BooleanSupplier verbose) {
     return this;
   }
 
+  /**
+   * Sets the consumer version.
+   *
+   * @return the builder with the version set
+   */
+  public ConsumerBuilder withVersion(String version) {
+    this.version = version;
+    return this;
+  }
+
   public Consumer build() {
     var clock = this.clock;
     if (clock == null) {
@@ -155,10 +166,16 @@ public Consumer build() {
     if (verbose == null) {
       verbose = () -> false;
     }
+    var version = this.version;
 
     Function<Stream<Grant>, JoinMessage> registerFactory =
         (grants) -> {
           var register = Messages.createRegisterBuilder(instance, service, grants);
+          if (version != null) {
+            var md = ClientMessage.Register.Metadata.newBuilder();
+            md.setVersion(version);
+            register.setMetadata(md);
+          }
           if (capacityLimit == null && keyNames == null) {
             return Messages.createRegister(register.build());
           }
@@ -173,7 +190,7 @@ public Consumer build() {
                 Stream.of(keyNames).map(Object::toString).collect(Collectors.joining(", ")));
             opts = opts.addAllNames(Stream.of(keyNames).map(DomainKeyName::toProto).toList());
           }
-          register = register.setOptions(opts.build());
+          register.setOptions(opts.build());
           return Messages.createRegister(register.build());
         };