问题标签 [hortonworks-dataflow]

问问题

For questions regarding programming in ECMAScript (JavaScript/JS) and its various dialects/implementations (excluding ActionScript). Note JavaScript is NOT the same as Java! Please include all relevant tags on your question; e.g., [node.js], [jquery], [json], [reactjs], [angular], [ember.js], [vue.js], [typescript], [svelte], etc.

66 问题

0 投票

1 回答

570 浏览

java - Hortonworks Schema Registry + Nifi + Java：反序列化 Nifi 记录

我正在尝试使用 Hortonworks Schema Registry 反序列化一些由 Nifi 序列化的 Kafka 消息

在 Nifi 端用作 RecordWritter 的处理器：AvroRecordSetWriter
Schema 写入策略：HWX CContent-Encoded Schema Reference

我能够在其他 Nifi kafka 消费者中反序列化这些消息。但是我正在尝试使用 Kafka 代码从我的 Flink 应用程序中反序列化它们。

我的 Flink 应用程序的 Kafka 反序列化处理程序中有以下内容：

这是反序列化消息的 HWXSchemaRegistryCode：

但我得到一个 404 HTTP 错误代码（找不到模式）。我认为这是由于 Nifi 配置和 HWX Schema Registry Client 实现之间的“协议”不兼容，因此客户端正在寻找的模式标识符字节在服务器上不存在，或者类似的东西。

有人可以帮忙吗？

谢谢你。

引起：javax.ws.rs.NotFoundException: HTTP 404 Not Found at org.glassfish.jersey.client.JerseyInvocation.convertToException(JerseyInvocation.java:1069) at org.glassfish.jersey.client.JerseyInvocation.translate(JerseyInvocation.java :866) 在 org.glassfish.jersey.client.JerseyInvocation.lambda$invoke$1(JerseyInvocation.java:750) 在 org.glassfish.jersey.internal.Errors.process(Errors.java:292) 在 org.glassfish.jersey .internal.Errors.process(Errors.java:274) 在 org.glassfish.jersey.internal.Errors.process(Errors.java:205) 在 org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java :390) 在 org.glassfish.jersey.client.JerseyInvocation.invoke(JerseyInvocation.java:748) 在 org.glassfish.jersey.client.JerseyInvocation$Builder。方法（JerseyInvocation.java:404）在 org.glassfish.jersey.client.JerseyInvocation$Builder.get(JerseyInvocation.java:300) 在 com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient$14.run(SchemaRegistryClient.java:1054 ) 在 com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient$14.run(SchemaRegistryClient.java:1051) 在 java.security.AccessController.doPrivileged(Native Method) 在 javax.security.auth.Subject.doAs(Subject.java: 360) 在 com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient.getEntities(SchemaRegistryClient.java:1051) 在 com.hortonworks.registries.schemaregistry.client.SchemaRegistryClient.getAllVersions(SchemaRegistryClient.java:872) 在 com.hortonworks.registries .schemaregistry.client.SchemaRegistryClient。getAllVersions(SchemaRegistryClient.java:676) 在 HWXSchemaRegistry.(HWXSchemaRegistry.java:56) 在 HWXSchemaRegistry.getInstance(HWXSchemaRegistry.java:26) 在 SchemaService.deserialize(SchemaService.java:70) 在 SchemaService.deserialize(SchemaService.java:26) ) 在 org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:140) 的 org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:45) ) 在 org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:93) 在 org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:712) 在 org. apache.flink.streaming.api.operators.StreamSource。在 org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask. java:302) 在 org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) 在 java.lang.Thread.run(Thread.java:745)