[Bug]: Error when using ReadFromCsv to read non-utf8 encoding file

[shunping]

What happened?

The following python code can be used to reproduce the problem.

"""Show an error when reading a non-utf8 encoding file with ReadFromCSV."""

import tempfile

import apache_beam as beam

from apache_beam.pipeline import Pipeline
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io import ReadFromCsv

def _run():
  content = b"\xe0,\xe1,\xe2\n0,1,2\n1,2,3\n2,3,4\n3,4,5\n4,5,6\n5,6,7\n" \
            b"6,7,8\n7,8,9\n8,9,0\n9,0,1\n0,1,2\n1,2,3\n"

  with tempfile.NamedTemporaryFile(suffix=".csv") as fp:
    input_fn = fp.name
    print(input_fn)
    with open(input_fn, 'wb') as f:
      f.write(content)

    pipeline_options = PipelineOptions()
    with Pipeline(options=pipeline_options) as p:
      _ = (
          p | 'Read' >> ReadFromCsv(input_fn, encoding="latin1")
          | 'CheckBytes' >> beam.Map(lambda x: print(x))
          )


if __name__ == "__main__":
  _run()

The error message is like UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe0 in position 0: unexpected end of data.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[shunping]

.take-issue