[BUG] java.lang.RuntimeException: Failed to create schema for topic

[daileizhi]

When I try to call dataset. Write(), output the data to pulsar, and throw the error that failed to create the schema

Environment:
Pulsar-2.4.1
Spark-2.4.4

java.lang.RuntimeException: Failed to create schema for persistent://public/default/als_statistics_tocheck
	at org.apache.spark.sql.pulsar.SchemaUtils$.uploadPulsarSchema(SchemaUtils.scala:104)
	at org.apache.spark.sql.pulsar.PulsarRowWriter.singleProducer$lzycompute(PulsarWriteTask.scala:140)
	at org.apache.spark.sql.pulsar.PulsarRowWriter.singleProducer(PulsarWriteTask.scala:138)
	at org.apache.spark.sql.pulsar.PulsarRowWriter.producerFlush(PulsarWriteTask.scala:210)

     RecordSchemaBuilder schemaBuilder = SchemaBuilder.record("topLevelRecord");
     schemaBuilder.field("ip").type(SchemaType.STRING);
     schemaBuilder.field("port").type(SchemaType.INT32);
     schemaBuilder.field("url_id").type(SchemaType.STRING);
     schemaBuilder.field("response_rate").type(SchemaType.DOUBLE);
     schemaBuilder.field("success_rate").type(SchemaType.DOUBLE);
     schemaBuilder.field("average_response_time").type(SchemaType.DOUBLE);
     schemaBuilder.field("average_network_time").type(SchemaType.DOUBLE);
     schemaBuilder.field("start_time").type(SchemaType.TIMESTAMP);
     schemaBuilder.field("end_time").type(SchemaType.TIMESTAMP);

     SchemaInfo  statistics2checkSchemaInfo = schemaBuilder.build(SchemaType.AVRO);

      waitToCheckDataSet.write()
                            //.mode("append")
                            .format("pulsar")
                            .option("service.url", serviceUrl)
                            .option("admin.url", adminUrl)
                            .option("topic", statistics2CheckTopic)
                            .option("pulsar.producer.sendTimeoutMs","60000")
                            //.option("avroSchema",statistics2checkSchemaInfo.getSchemaDefinition())
                            //.option("recordName","CheckDataSet")
                            //.option("recordNamespace","com.some.domain")
                            .save();

    try {
            admin.schemas().getSchemaInfo(statistics2CheckTopic);
        } catch (PulsarAdminException e) {
            if (404 == e.getStatusCode()) {
                admin.schemas().createSchema(statistics2CheckTopic,statistics2checkSchemaInfo);
            }
        }

        spark.readStream()
                .format("pulsar")
                .option("service.url", serviceUrl)
                .option("admin.url", adminUrl)
                .option("topic", statistics2CheckTopic)
                .option("startingOffsets", "earliest")
                .load()
                .withWatermark("__eventTime", "1 minute")
                .writeStream().queryName("WaitToCheckDataSet")
                .outputMode("append")
                .trigger(Trigger.ProcessingTime("1 minute"))
                .foreachBatch((dataset,batchId) -> {
                    System.out.println("------WaitToCheckDataSet-------");
                     dataset.show(false);
               }).start();

In addition,how can support options "avroSchema,recordName,recordNamespace"
The name of the schema has always been named "topLevelRecord".Except : com.some.domain.

[http://spark.apache.org/docs/latest/sql-data-sources-avro.html](url)
{
    "type": "record",
    "name": "topLevelRecord",
    "fields": [
      {
        "name": "ip",
        "type": [
          "string",
          "null"
        ]
      },
      {
        "name": "port",
        "type": [
          "int",
          "null"
        ]
      },
      {
        "name": "url_id",
        "type": "string"
      },
      {
        "name": "response_rate",
        "type": [
          "double",
          "null"
        ]
      },
      {
        "name": "success_rate",
        "type": [
          "double",
          "null"
        ]
      },
      {
        "name": "average_response_time",
        "type": "double"
      },
      {
        "name": "average_network_time",
        "type": "double"
      },
      {
        "name": "start_time",
        "type": [
          {
            "type": "long",
            "logicalType": "timestamp-micros"
          },
          "null"
        ]
      },
      {
        "name": "end_time",
        "type": [
          {
            "type": "long",
            "logicalType": "timestamp-micros"
          },
          "null"
        ]
      },
      {
        "name": "type",
        "type": "int"
      }
    ]
  }

[swapnil-chougule]

Any resolution for this issue ?

[syhily]

@swapnil-chougule Can you paste your exception trace here?

[syhily]

SchemaUtils$.uploadPulsarSchema(SchemaUtils.scala:104) indicate that you should have admin privilege on the topic. This should be the root cause for this exception. So I don't think this is a bug.