title: Confluent HDFS Sink Connector : Kafka Topic with plain string format to HDFS in Parquet format fails with Avro schema must be a record error
body: <p>I have installed docker confluentinc/cp-kafka-connect:4.0.0 image in my virtual machine. I am interested in getting the kafka topic which is a plain text data (string format) in hdfs in parquet format.</p>
  
  <pre><code>sudo docker run -d \
      --name=kafka-connect \
      --net=host \
      -e CONNECT_BOOTSTRAP_SERVERS=&lt;kafka-server-host&gt;:9092 \
      -e CONNECT_REST_PORT=8082 \
      -e CONNECT_GROUP_ID="connect-kafkac1" \
      -e CONNECT_CONFIG_STORAGE_TOPIC="connect-kafkac1-config" \
      -e CONNECT_OFFSET_STORAGE_TOPIC="connect-kafkac1-offsets" \
      -e CONNECT_STATUS_STORAGE_TOPIC="connect-kafkac1-status" \
      -e CONNECT_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
      -e CONNECT_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
      -e CONNECT_INTERNAL_KEY_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
      -e CONNECT_INTERNAL_VALUE_CONVERTER="org.apache.kafka.connect.json.JsonConverter" \
      -e CONNECT_REST_ADVERTISED_HOST_NAME="localhost" \
      -e CONNECT_LOG4J_LOGGERS=org.reflections=ERROR \
      -e CONNECT_PLUGIN_PATH=/usr/share/java \
      confluentinc/cp-kafka-connect:4.0.0
  </code></pre>
  
  <p>I have the below configurations done.</p>
  
  <p><strong><em>/etc/kafka/connect-standalone.properties</em></strong></p>
  
  <pre><code>bootstrap.servers=kafkaclusteraddress:9092
  key.converter=org.apache.kafka.connect.storage.StringConverter
  value.converter=org.apache.kafka.connect.storage.StringConverter
  key.converter.schemas.enable=false
  value.converter.schemas.enable=false
  internal.key.converter=org.apache.kafka.connect.json.JsonConverter
  internal.value.converter=org.apache.kafka.connect.json.JsonConverter
  internal.key.converter.schemas.enable=false
  internal.value.converter.schemas.enable=false
  </code></pre>
  
  <p><strong><em>/etc/kafka-connect-hdfs/quickstart-hdfs.properties</em></strong></p>
  
  <pre><code>name=hdfs-sink
  connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
  tasks.max=1
  topics=&lt;topicname&gt;
  topics.dir=&lt;hdfs-topic-dir&gt;
  logs.dir=&lt;hdfs-logs-dir&gt;
  hdfs.url=&lt;hdfs-url&gt;:8020
  format.class=io.confluent.connect.hdfs.parquet.ParquetFormat
  flush.size=3
  hadoop.conf.dir=/etc/hadoop/conf
  hadoop.home=/usr/bin/hadoop
  </code></pre>
  
  <p><strong><em>error i receive after running the config in standalone mode</em></strong></p>
  
  <pre><code># connect-standalone /etc/kafka/connect-standalone.properties /etc/kafka-connect-hdfs/quickstart-hdfs.properties
  
  ----------------------------------------
  
  [2018-02-13 19:11:09,542] ERROR WorkerSinkTask{id=hdfs-sink-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. (org.apache.kafka.connect.runtime.WorkerSinkTask)
  java.lang.IllegalArgumentException: Avro schema must be a record.
          at org.apache.parquet.avro.AvroSchemaConverter.convert(AvroSchemaConverter.java:113)
          at org.apache.parquet.avro.AvroParquetWriter.writeSupport(AvroParquetWriter.java:144)
          at org.apache.parquet.avro.AvroParquetWriter.&lt;init&gt;(AvroParquetWriter.java:106)
          at io.confluent.connect.hdfs.parquet.ParquetRecordWriterProvider$1.write(ParquetRecordWriterProvider.java:68)
          at io.confluent.connect.hdfs.TopicPartitionWriter.writeRecord(TopicPartitionWriter.java:635)
          at io.confluent.connect.hdfs.TopicPartitionWriter.write(TopicPartitionWriter.java:379)
          at io.confluent.connect.hdfs.DataWriter.write(DataWriter.java:374)
          at io.confluent.connect.hdfs.HdfsSinkTask.put(HdfsSinkTask.java:101)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:495)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:288)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:198)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:166)
          at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
          at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
          at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
          at java.util.concurrent.FutureTask.run(FutureTask.java:266)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          at java.lang.Thread.run(Thread.java:745)
  [2018-02-13 19:11:09,545] ERROR WorkerSinkTask{id=hdfs-sink-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask)
  org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
          at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:517)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:288)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:198)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:166)
          at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:170)
          at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:214)
          at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
          at java.util.concurrent.FutureTask.run(FutureTask.java:266)
          at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          at java.lang.Thread.run(Thread.java:745)
  Caused by: java.lang.IllegalArgumentException: Avro schema must be a record.
          at org.apache.parquet.avro.AvroSchemaConverter.convert(AvroSchemaConverter.java:113)
          at org.apache.parquet.avro.AvroParquetWriter.writeSupport(AvroParquetWriter.java:144)
          at org.apache.parquet.avro.AvroParquetWriter.&lt;init&gt;(AvroParquetWriter.java:106)
          at io.confluent.connect.hdfs.parquet.ParquetRecordWriterProvider$1.write(ParquetRecordWriterProvider.java:68)
          at io.confluent.connect.hdfs.TopicPartitionWriter.writeRecord(TopicPartitionWriter.java:635)
          at io.confluent.connect.hdfs.TopicPartitionWriter.write(TopicPartitionWriter.java:379)
          at io.confluent.connect.hdfs.DataWriter.write(DataWriter.java:374)
          at io.confluent.connect.hdfs.HdfsSinkTask.put(HdfsSinkTask.java:101)
          at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:495)
          ... 10 more
  [2018-02-13 19:11:09,545] ERROR WorkerSinkTask{id=hdfs-sink-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask)
  </code></pre>