apache_beam.io.external.kafka.ReadFromKafka throws IndexError

Kafka.ReadFromKafka throws *IndexError: tuple index out of range* due to unimplemented "*_get_named_tuple_instance*" function  of class *SchemaBasedPayloadBuilder(PayloadBuilder):*

*Stacktrace:*

Traceback (most recent call last):
File "/usr/lib/python3.9/runpy.py", line 197, in _run_module_as_main
return _run_code(code, main_globals, None,
File "/usr/lib/python3.9/runpy.py", line 87, in _run_code
exec(code, run_globals)
File "/code/src/beam_example/beamKafkaRedis.py", line 36, in
notifications = pipeline | "Reading messages from Kafka" \>\> kafka.ReadFromKafka(
File "/usr/local/lib/python3.9/dist-packages/apache_beam/io/kafka.py", line 166, in __init__
super(ReadFromKafka, self).__init__(
File "/usr/local/lib/python3.9/dist-packages/apache_beam/transforms/external.py", line 217, in __init__
payload.payload() if isinstance(payload, PayloadBuilder) else payload)
File "/usr/local/lib/python3.9/dist-packages/apache_beam/transforms/external.py", line 93, in payload
return self.build().SerializeToString()
File "/usr/local/lib/python3.9/dist-packages/apache_beam/transforms/external.py", line 106, in build
schema = named_tuple_to_schema(type(row))
File "/usr/local/lib/python3.9/dist-packages/apache_beam/typehints/schemas.py", line 276, in named_tuple_to_schema
return typing_to_runner_api(named_tuple).row_type.schema
File "/usr/local/lib/python3.9/dist-packages/apache_beam/typehints/schemas.py", line 184, in typing_to_runner_api
element_type = typing_to_runner_api(_get_args(type_)[0])
IndexError: tuple index out of range
args: ['\--runner=PortableRunner', '\--streaming', '\--sdk_worker_parallelism=2', '\--job_name=beam-readKafkaTopic', '\--environment_type=PROCESS', '\--environment_config=\{"command": "/opt/apache/beam/boot"}', '\--job_name=beam-kafkaConnect', '\--job_endpoint=localhost:39295']

at org.apache.beam.runners.flink.FlinkPortableClientEntryPoint.runDriverProgram(FlinkPortableClientEntryPoint.java:192) ~[?:?]
at org.apache.beam.runners.flink.FlinkPortableClientEntryPoint.main(FlinkPortableClientEntryPoint.java:100) ~[?:?]
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:?]
at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
at java.lang.reflect.Method.invoke(Method.java:566) ~[?:?]
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355) ~[flink-dist_2.12-1.13.2.jar:1.13.2]

Imported from Jira [BEAM-12848](https://issues.apache.org/jira/browse/BEAM-12848). Original Jira may contain additional context.
Reported by: Harsh_99.

Start in apache_beam/io/kafka.py and follow the external payload path through transforms/external.py into typehints/schemas.py, especially named_tuple_to_schema and the failing type argument lookup. Reproduce the ReadFromKafka construction shown in the stack trace and trace why the schema has no indexable type arguments. Done means ReadFromKafka no longer raises this IndexError for the reported configuration and the behavior is covered by an appropriate regression check.

Written by the indexing model from the issue text.