add offset type to NewOffset func (#962)

* add tests for explicit offset type

* add Offset type case and return

Co-authored-by: Milind L <[email protected]>

---------

Co-authored-by: Jack Dockerty <[email protected]>
Co-authored-by: Confluent Jenkins Bot <[email protected]>
Co-authored-by: Milind L <[email protected]>

CHANGELOG.md

@@ -37,6 +37,7 @@ This is a feature release:
   * Added `SetRoundtripDuration` to the mock broker for setting RTT delay for
     a given mock broker (@kkoehler, #892).
   * Built-in support for Linux/ arm64. (#933).
+  * Support for Offset types on `Offset.Set()`
 
 ### Fixes
 

kafka/kafka_test.go

@@ -131,6 +131,22 @@ func TestOffsetAPIs(t *testing.T) {
 		t.Errorf("Failed to change offset. Expect (%v), got (%v)\n", 10, offset)
 	}
 
+	// test passing explicit 'Offset' type
+	err = offset.Set(OffsetBeginning)
+	if err != nil {
+		t.Errorf("Cannot set offset to (%v). Error: %s \n", OffsetBeginning, err)
+	} else if offset != OffsetBeginning {
+		t.Errorf("Failed to change offset. Expect (%v), got %v\n", OffsetBeginning, offset)
+	}
+
+	// test passing explicit 'Offset' type
+	err = offset.Set(OffsetEnd)
+	if err != nil {
+		t.Errorf("Cannot set offset to (%v). Error: %s \n", OffsetEnd, err)
+	} else if offset != OffsetEnd {
+		t.Errorf("Failed to change offset. Expect (%v), got %v\n", OffsetEnd, offset)
+	}
+
 	// test OffsetTail()
 	tail := OffsetTail(offset)
 	t.Logf("offset tail %v\n", tail)

kafka/offset.go

@@ -72,8 +72,8 @@ func (o *Offset) Set(offset interface{}) error {
 	return err
 }
 
-// NewOffset creates a new Offset using the provided logical string, or an
-// absolute int64 offset value.
+// NewOffset creates a new Offset using the provided logical string, an
+// absolute int64 offset value, or a concrete Offset type.
 // Logical offsets: "beginning", "earliest", "end", "latest", "unset", "invalid", "stored"
 func NewOffset(offset interface{}) (Offset, error) {
 
@@ -107,6 +107,8 @@ func NewOffset(offset interface{}) (Offset, error) {
 		return Offset((int64)(v)), nil
 	case int64:
 		return Offset(v), nil
+	case Offset:
+		return Offset(v), nil
 	default:
 		return OffsetInvalid, newErrorFromString(ErrInvalidArg,
 			fmt.Sprintf("Invalid offset type: %t", v))