Disable KafkaIO SDF while it is tested and fixed

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java

@@ -543,6 +543,11 @@
  * For any significant significant updates to this I/O connector, please consider involving
  * corresponding code reviewers mentioned <a
  * href="https://github.com/apache/beam/blob/master/sdks/java/io/kafka/OWNERS">here</a>.
+ *
+ * <h1>Reading from Kafka SDF is currently broken, as re-starting the pipeline will cause the
+ * consumer to start from scratch. See <a
+ * href="https://github.com/apache/beam/issues/21730">this</a>. Current workaround is to use
+ * --experimental_option=use_unbounded_sdf_wrapper to use the Unbounded implementation</h1>
  */
 @Experimental(Kind.SOURCE_SINK)
 @SuppressWarnings({
@@ -1357,6 +1362,10 @@ public PCollection<KafkaRecord<K, V>> expand(PBegin input) {
               && runnerPrefersLegacyRead(input.getPipeline().getOptions()))) {
         return input.apply(new ReadFromKafkaViaUnbounded<>(this, keyCoder, valueCoder));
       }
+      LOG.warn(
+          "Reading from Kafka SDF is currently broken, as re-starting the pipeline will cause the consumer to start from scratch."
+              + " See https://github.com/apache/beam/issues/21730 . "
+              + "Current workaround is to use --experimental_option=use_unbounded_sdf_wrapper to use the Unbounded implementation");
       return input.apply(new ReadFromKafkaViaSDF<>(this, keyCoder, valueCoder));
     }
 

sdks/python/apache_beam/io/kafka.py

@@ -76,16 +76,33 @@
 
   For more information specific to Flink runner see:
   - https://beam.apache.org/documentation/runners/flink/
+
+  Reading via Kafka SDF is currently broken, and will cause the pipeline
+  to re-read all data on Kafka whenever restarted.
+  See https://github.com/apache/beam/issues/21730.
+  Current workaround is:
+  Start an expansion service with experiment "use_unbounded_sdf_wrapper"
+  #pylint: disable=line-too-long
+  java -jar sdks/java/io/expansion-service/build/libs/beam-sdks-java-io-expansion-service-2.41.0-SNAPSHOT.jar 12345 --experiments=use_unbounded_sdf_wrapper
+  #pylint: enable=line-too-long
+
+  Update transforms in kafka.py to use this expansion service.
+  https://github.com/apache/beam/blob/2c8e7eb7a39cbe3a1678a5c6b8b3f8700d4d8706/sdks/python/apache_beam/io/kafka.py#L189
+
+  Use instructions in Kafka example to run this from a Git clone.
+  https://github.com/apache/beam/tree/master/sdks/python/apache_beam/examples/kafkataxi
 """
 
 # pytype: skip-file
-
+import logging
 import typing
 
 from apache_beam.transforms.external import BeamJarExpansionService
 from apache_beam.transforms.external import ExternalTransform
 from apache_beam.transforms.external import NamedTupleBasedPayloadBuilder
 
+_LOGGER = logging.getLogger(__name__)
+
 ReadFromKafkaSchema = typing.NamedTuple(
     'ReadFromKafkaSchema',
     [('consumer_config', typing.Mapping[str, str]),
@@ -166,6 +183,10 @@ def __init__(
         Java Kafka Reader reads keys and values as 'byte[]'.
     :param expansion_service: The address (host:port) of the ExpansionService.
     """
+    _LOGGER.warning(
+        "Reading from Kafka via SDF is currently broken, and the pipeline will"
+        "re-read all data on the topic whenever the pipeline is restarted. "
+        "See Kafka.py for workaround instructions.")
     if timestamp_policy not in [ReadFromKafka.processing_time_policy,
                                 ReadFromKafka.create_time_policy,
                                 ReadFromKafka.log_append_time]: