From 4aee33d6ab1345243e426e05388f6fc512970e93 Mon Sep 17 00:00:00 2001
From: TengYao Chi <kitingiao@gmail.com>
Date: Tue, 17 Dec 2024 00:20:30 +0800
Subject: [PATCH] KAFKA-18259: Documentation for consumer auto.offset.reset
 contains invalid HTML (#18210)

Reviewers: Andrew Schofield <aschofield@confluent.io>
---
 .../kafka/clients/consumer/ConsumerConfig.java      |  4 ++--
 .../consumer/internals/AutoOffsetResetStrategy.java | 13 +++++++++++++
 2 files changed, 15 insertions(+), 2 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index fb4c05062bb9f..b0b3927407593 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -171,9 +171,9 @@ public class ConsumerConfig extends AbstractConfig {
     public static final String AUTO_OFFSET_RESET_CONFIG = "auto.offset.reset";
     public static final String AUTO_OFFSET_RESET_DOC = "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server " +
             "(e.g. because that data has been deleted): " +
-            "<ul><li>earliest: automatically reset the offset to the earliest offset" +
+            "<ul><li>earliest: automatically reset the offset to the earliest offset</li>" +
             "<li>latest: automatically reset the offset to the latest offset</li>" +
-            "<li>by_duration:<duration>: automatically reset the offset to a configured <duration> from the current timestamp. <duration> must be specified in ISO8601 format (PnDTnHnMn.nS). " +
+            "<li>by_duration:&lt;duration&gt;: automatically reset the offset to a configured &lt;duration&gt; from the current timestamp. &lt;duration&gt; must be specified in ISO8601 format (PnDTnHnMn.nS). " +
             "Negative duration is not allowed.</li>" +
             "<li>none: throw exception to the consumer if no previous offset is found for the consumer's group</li>" +
             "<li>anything else: throw exception to the consumer.</li></ul>" +
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java
index a692bc45ea7bd..e904ca3d5d63a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java
@@ -27,6 +27,7 @@
 import java.util.Locale;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 public class AutoOffsetResetStrategy {
     public enum StrategyType {
@@ -165,5 +166,17 @@ public void ensureValid(String name, Object value) {
                         name + ". The value must be either 'earliest', 'latest', 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");
             }
         }
+
+        @Override
+        public String toString() {
+            String values = Arrays.stream(StrategyType.values())
+                .map(strategyType -> {
+                    if (strategyType == StrategyType.BY_DURATION) {
+                        return "by_duration:PnDTnHnMn.nS";
+                    }
+                    return strategyType.toString();
+                }).collect(Collectors.joining(", "));
+            return "[" + values + "]";
+        }
     }
 }