使用经过身份验证的ConFluent Schema注册表配置Spark结构化流

Question

我在Spark流中使用Kafka源来接收在Confluent Cloud中使用Datagen生成的记录。我打算使用融合模式注册表，

目前，这是我面临的例外情况:*

线程"main"io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException中的异常：未经授权；错误代码：401

融合云的模式注册表需要传递一些我不知道如何输入的身份验证数据：

basic.auth.credentials.source=USER_INFO
schema.registry.basic.auth.user.info=secret: secret

我认为我必须将此身份验证数据传递给CachedSchemareGistrCyclient，但我不确定是否传递，以及如何传递。

// Setup the Avro deserialization UDF
   schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 128)
    kafkaAvroDeserializer = new AvroDeserializer(schemaRegistryClient)
    spark.udf.register("deserialize", (bytes: Array[Byte]) =>
      kafkaAvroDeserializer.deserialize(bytes)

如果我试图以身份向模式注册表发送身份验证

val restService = new RestService(schemaRegistryURL)

  val props = Map(
    "basic.auth.credentials.source" -> "USER_INFO",
    "schema.registry.basic.auth.user.info" -> "secret:secret"
  ).asJava

  var schemaRegistryClient = new CachedSchemaRegistryClient(restService, 100, props)

我得到< code >无法解析重载的构造函数CachedSchemaRegistryClient ，似乎只有2个参数要发送到CachedSchemaRegistryClient。

如何解决此问题？

我遇到了这篇文章，但在这里他们没有对融合云中的模式注册表应用任何身份验证。

Answer 1

这段代码对我有用:

private val schemaRegistryUrl = "<schemaregistryURL>"   
val props = Map("basic.auth.credentials.source" -> "USER_INFO",
 "schema.registry.basic.auth.user.info" -> "<api-key>:<api-secret>").asJava

 private val schemaRegistryClient = new CachedSchemaRegistryClient(schemaRegistryUrl, 100,props)

我们需要确保在转换为JAVA时进行正确的导入：

 import scala.collection.JavaConverters.mapAsJavaMapConverter