From 406d0e048172b8e1cfab2b6257327f3cb487c5b6 Mon Sep 17 00:00:00 2001
From: pastor <viruszold@gmail.com>
Date: Mon, 21 Sep 2020 16:19:54 +0300
Subject: [PATCH] Write balance port

---
 .../java/io/confluent/kafkarest/KafkaRestConfig.java | 12 ++++++++++++
 .../main/java/io/confluent/kafkarest/UriUtils.java   |  5 ++++-
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/kafka-rest/src/main/java/io/confluent/kafkarest/KafkaRestConfig.java b/kafka-rest/src/main/java/io/confluent/kafkarest/KafkaRestConfig.java
index 6649d161bb..e9a48d570e 100644
--- a/kafka-rest/src/main/java/io/confluent/kafkarest/KafkaRestConfig.java
+++ b/kafka-rest/src/main/java/io/confluent/kafkarest/KafkaRestConfig.java
@@ -74,6 +74,12 @@ public class KafkaRestConfig extends RestConfig {
       + " hostname is used";
   public static final String HOST_NAME_DEFAULT = "";
 
+  public static final String HOST_PORT_CONFIG = "host.port";
+  private static final String HOST_PORT_DOC =
+      "The host port used to generate absolute URLs in responses. If empty, the default canonical"
+          + " port is used";
+  public static final String HOST_PORT_DEFAULT = "-1";
+
   public static final String ADVERTISED_LISTENERS_CONFIG = "advertised.listeners";
   protected static final String ADVERTISED_LISTENERS_DOC =
       "List of advertised listeners. Used when generating absolute URLs in responses. Protocols"
@@ -376,6 +382,12 @@ protected static ConfigDef baseKafkaRestConfigDef() {
         HOST_NAME_DEFAULT,
         Importance.MEDIUM, HOST_NAME_DOC
     )
+    .define(
+        HOST_PORT_CONFIG,
+        Type.INT,
+        HOST_PORT_DEFAULT,
+        Importance.MEDIUM, HOST_PORT_DOC
+    )
     .define(
         ADVERTISED_LISTENERS_CONFIG,
         Type.LIST,
diff --git a/kafka-rest/src/main/java/io/confluent/kafkarest/UriUtils.java b/kafka-rest/src/main/java/io/confluent/kafkarest/UriUtils.java
index 6f3ef6aca2..a02df6ccc0 100644
--- a/kafka-rest/src/main/java/io/confluent/kafkarest/UriUtils.java
+++ b/kafka-rest/src/main/java/io/confluent/kafkarest/UriUtils.java
@@ -35,7 +35,10 @@ public static UriBuilder absoluteUriBuilder(KafkaRestConfig config, UriInfo uriI
       URI origAbsoluteUri = uriInfo.getAbsolutePath();
       builder.scheme(origAbsoluteUri.getScheme());
       // Only reset the port if it was set in the original URI
-      if (origAbsoluteUri.getPort() != -1) {
+      Integer port = config.getInt(KafkaRestConfig.HOST_PORT_CONFIG);
+      if (port != null && port != -1) {
+        builder.port(port);
+      } else if (origAbsoluteUri.getPort() != -1) {
         try {
           builder.port(config.consumerPort(origAbsoluteUri.getScheme()));
         } catch (URISyntaxException e) {