Address Gemini review comment & update CHANGES.md

Author: tkaymak

CHANGES.md

@@ -65,6 +65,7 @@
 ## I/Os
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* DebeziumIO (Java): added `OffsetRetainer` interface and `FileSystemOffsetRetainer` implementation to persist and restore CDC offsets across pipeline restarts, and exposed `withStartOffset` / `withOffsetRetainer` on `DebeziumIO.Read` and the cross-language `ReadBuilder` ([#28248](https://github.com/apache/beam/issues/28248)).
 
 ## New Features / Improvements
 

sdks/java/io/debezium/src/main/java/org/apache/beam/io/debezium/DebeziumTransformRegistrar.java

@@ -138,6 +138,12 @@ public PTransform<PBegin, PCollection<String>> buildExternal(Configuration confi
         Map<String, Object> startOffsetMap = new HashMap<>();
         for (String property : configuration.startOffset) {
           String[] parts = property.split("=", 2);
+          if (parts.length != 2) {
+            throw new IllegalArgumentException(
+                "Invalid startOffset entry: \""
+                    + property
+                    + "\". Expected format is \"key=value\".");
+          }
           String key = parts[0];
           String value = parts[1];
           try {

sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOTest.java

@@ -27,6 +27,7 @@
 import io.debezium.connector.mysql.MySqlConnector;
 import io.debezium.connector.mysql.MySqlConnectorConfig;
 import java.io.Serializable;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.beam.io.debezium.DebeziumIO.ConnectorConfiguration;
@@ -231,4 +232,19 @@ public void testRetainerFallsBackToWithStartOffsetWhenLoadReturnsNull() throws E
     KafkaSourceConsumerFn.OffsetHolder restriction = fn.getInitialRestriction(null);
     assertEquals(explicitOffset, restriction.offset);
   }
+
+  @Test
+  public void testBuildExternalThrowsOnMalformedStartOffsetEntry() {
+    DebeziumTransformRegistrar.ReadBuilder.Configuration config =
+        new DebeziumTransformRegistrar.ReadBuilder.Configuration();
+    config.setUsername("user");
+    config.setPassword("pass");
+    config.setHost("localhost");
+    config.setPort("3306");
+    config.setConnectorClass("MySQL");
+    config.setStartOffset(Arrays.asList("lsn=100", "no-equals-sign"));
+    assertThrows(
+        IllegalArgumentException.class,
+        () -> new DebeziumTransformRegistrar.ReadBuilder().buildExternal(config));
+  }
 }