java - KStream-KStream 内连接抛出 java.lang.ClassCastException
问题描述
在 的流程方法中@StreamListener
,我将学校 KStream 映射到人员 KStream 并通过 .through() 方法填充主题“人员”,从该主题“人员”中生成KStream
另一个 process1 方法
@StreamListener
。
MianApplication.java
@SpringBootApplication
public class KafkaStreamsTableJoin {
public static void main(String[] args) {
SpringApplication.run(KafkaStreamsTableJoin.class, args);
}
@EnableBinding(KStreamProcessorX.class)
public static class KStreamToTableJoinApplication {
@StreamListener
public void process(@Input("school") KStream<SchoolKey, School> schools) {
schools.map((schoolKey, school) -> {
return KeyValue.pair(new PersonKey("Adam", "Smith", schoolKey.getId()), new Person(12));
})
.through("person", Produced.with(new PersonKeySerde(), new PersonSerde()));
}
@StreamListener
public void process1(@Input("school_1") KStream<SchoolKey, School> schools, @Input("person") KStream<PersonKey, Person> persons) {
schools.selectKey((schoolKey, school) -> schoolKey.getId())
.join(persons.selectKey((personKey, person) -> personKey.getId()),
(school, person) -> {
System.out.println("school_app2= " + school + ", person_app2= " + person);
return null;
},
JoinWindows.of(Duration.ofSeconds(1)),
Joined.with(Serdes.Integer(), new SchoolSerde(), new PersonSerde())
);
}
}
interface KStreamProcessorX {
@Input("person")
KStream<?, ?> inputPersonKStream();
@Input("school")
KStream<?, ?> inputSchoolKStream();
@Input("school_1")
KStream<?, ?> inputSchool1KStream();
}
}
在方法 process1 中,此 KStream 需要与另一个 KStream 加入,但出现以下异常:
Exception in thread "stream-join-sample_2-654e8060-5b29-4694-9188-032a9779529c-StreamThread-1" java.lang.ClassCastException: class kafka.streams.join.School cannot be cast to class kafka.streams.join.Person (kafka.streams.join.School and kafka.streams.join.Person are in unnamed module of loader 'app')
at org.apache.kafka.streams.kstream.internals.AbstractStream.lambda$reverseJoiner$0(AbstractStream.java:98)
at org.apache.kafka.streams.kstream.internals.KStreamKStreamJoin$KStreamKStreamJoinProcessor.process(KStreamKStreamJoin.java:94)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
at org.apache.kafka.streams.kstream.internals.KStreamJoinWindow$KStreamJoinWindowProcessor.process(KStreamJoinWindow.java:55)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:366)
at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:889)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:804)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:773)
我认为这个异常与不正确的 serde 有关,但我不知道是哪个 serde 造成了问题以及如何解决它。还是在方法过程中的映射过程中,触发了重新分区,这与不正确的serde有关?
POJO 和 Serde 的:
人.java
public class Person {
private double age;
public Person() {
}
public Person(double age) {
this.age = age;
}
@JsonGetter("age")
public double getAge() {
return age;
}
@JsonSetter("age")
public void setAge(double age) {
this.age = age;
}
@Override
public String toString() {
return "Person{" +
"age=" + age +
'}';
}
}
PersonSerde.java
public class PersonSerde extends Serdes.WrapperSerde<Person> {
public PersonSerde () {
super(new JsonSerializer<>(), new JsonDeserializer<>(Person.class));
}
}
个人密钥.java
public class PersonKey {
private String firstName;
private String lastName;
private int id;
public PersonKey() {
}
public PersonKey(String firstName, String lastName, int id) {
this.firstName = firstName;
this.lastName = lastName;
this.id = id;
}
@JsonGetter("firstName")
public String getFirstName() {
return firstName;
}
@JsonSetter("firstName")
public void setFirstName(String firstName) {
this.firstName = firstName;
}
@JsonGetter("lastName")
public String getLastName() {
return lastName;
}
@JsonSetter("lastName")
public void setLastName(String lastName) {
this.lastName = lastName;
}
@JsonGetter("id")
public int getId() {
return id;
}
@JsonSetter("id")
public void setId(int id) {
this.id = id;
}
@Override
public String toString() {
return "PersonKey{" +
"firstName='" + firstName + '\'' +
", lastName='" + lastName + '\'' +
", id=" + id +
'}';
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
PersonKey personKey = (PersonKey) o;
return id == personKey.id &&
Objects.equals(firstName, personKey.firstName) &&
Objects.equals(lastName, personKey.lastName);
}
@Override
public int hashCode() {
return Objects.hash(firstName, lastName, id);
}
}
PersonKeySerde.java
public class PersonKeySerde extends Serdes.WrapperSerde<PersonKey> {
public PersonKeySerde () {
super(new JsonSerializer<>(), new JsonDeserializer<>(PersonKey.class));
}
}
School 类的 serde 和 pojo 类似于 Person 类。
应用程序.yml
spring.application.name: stream-join-sample
spring.cloud.stream.bindings.school:
destination: school
contentType: application/json
consumer:
useNativeDecoding: false
spring.cloud.stream.kafka.streams.bindings.school:
consumer:
keySerde: kafka.streams.serde.SchoolKeySerde
valueSerde: kafka.streams.serde.SchoolSerde
application-id: stream-join-sample_1
spring.cloud.stream.bindings.person:
destination: person
contentType: application/json
consumer:
useNativeDecoding: false
spring.cloud.stream.kafka.streams.bindings.person:
consumer:
keySerde: kafka.streams.serde.PersonKeySerde
valueSerde: kafka.streams.serde.PersonSerde
application-id: stream-join-sample_2
spring.cloud.stream.bindings.school_1:
destination: school
contentType: application/json
consumer:
useNativeDecoding: false
spring.cloud.stream.kafka.streams.bindings.school_1:
consumer:
keySerde: kafka.streams.serde.SchoolKeySerde
valueSerde: kafka.streams.serde.SchoolSerde
application-id: stream-join-sample_2
spring.cloud.stream.kafka.streams.binder:
brokers: localhost
configuration:
default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
commit.interval.ms: 100
具有可重复步骤的示例应用程序:
解决方案
我从 GitHub 下载了您的代码以深入研究,结果发现它实际上是使用的JsonSerializer
/中的一个错误JsonDeserializer
。类型 ( School
, Person
, PersonKey
, SchoolKey
) 在记录标头中编码,但标头永远不会被清除。每次类型更改时,只会附加一个新的标头(标头键不是唯一的,并且允许重复)。
对于某些记录,同一类型只是被多次编码,因此这部分代码有效。但是,在某些情况下,会编码不同的类型,并且在从主题中读取数据时会“随机”选择一种类型(找到的第一个标头)。发生在连接之前,但在从重新分区主题接收数据时。如果选择了错误的类型,则代码ClassCastException
稍后会崩溃。
新答案:
在对这张票的讨论之后,https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/685,您应该通过以下方式禁用将类型信息写入记录标头:
props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
请注意,所有Serdes
手动创建的,即通过调用new
必须手动配置:
Map<String, Object> config = new HashMap<>();
config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
PersonKeySerde personKeySerde = new PersonKeySerde();
personKeySerde.configure(config, true);
PersonSerde personSerde = new PersonSerde();
personSerde.configure(config, false);
// ...
.through("person", Produced.with(personKeySerde, personSerde));
原答案:
作为一种解决方法,您可以将map
和替换selectKey()
为 atransform()
并清除transform()
. 虽然这是一个黑客。您应该向 SpringBoot 项目提交一张票,以便他们可以修复JsonSerializer
/ JsonDeserializer
。
以下代码删除标题并确保使用正确的类型,避免ClassCastException
:
@SpringBootApplication
public class KafkaStreamJoinApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaStreamJoinApplication.class, args);
}
@EnableBinding(KStreamProcessorX.class)
public static class KafkaKStreamJoinApplication {
@StreamListener
public void process(@Input("school") KStream<SchoolKey, School> schools) {
// replace map() with transform()
schools.transform(new TransformerSupplier<SchoolKey, School, KeyValue<PersonKey, Person>>() {
@Override
public Transformer<SchoolKey, School, KeyValue<PersonKey, Person>> get() {
return new Transformer<SchoolKey, School, KeyValue<PersonKey, Person>>() {
ProcessorContext context;
@Override
public void init(final ProcessorContext context) {
this.context = context;
}
@Override
public KeyValue<PersonKey, Person> transform(final SchoolKey key, final School value) {
// clear all headers; would be sufficient to only remove type header
for (Header h : context.headers().toArray()) {
context.headers().remove(h.key());
}
// same a "old" map code:
return KeyValue.pair(new PersonKey("Adam", "Smith", key.getId()), new Person(12));
}
@Override
public void close() {}
};
}})
.through("person", Produced.with(new PersonKeySerde(), new PersonSerde()));
}
@StreamListener
public void process1(@Input("school_1") KStream<SchoolKey, School> schools, @Input("person") KStream<PersonKey, Person> persons) {
// replace selectKey() with transform()
schools.transform(new TransformerSupplier<SchoolKey, School, KeyValue<Integer, School>>() {
@Override
public Transformer<SchoolKey, School, KeyValue<Integer, School>> get() {
return new Transformer<SchoolKey, School, KeyValue<Integer, School>>() {
ProcessorContext context;
@Override
public void init(final ProcessorContext context) {
this.context = context;
}
@Override
public KeyValue<Integer, School> transform(final SchoolKey key, final School value) {
// clear all headers; would be sufficient to only remove type header
for (Header h : context.headers().toArray()) {
context.headers().remove(h.key());
}
// effectively the same as "old" selectKey code:
return KeyValue.pair(key.getId(), value);
}
@Override
public void close() {}
};
}})
// replace selectKey() with transform()
.join(persons.transform(new TransformerSupplier<PersonKey, Person, KeyValue<Integer, Person>>() {
@Override
public Transformer<PersonKey, Person, KeyValue<Integer, Person>> get() {
return new Transformer<PersonKey, Person, KeyValue<Integer, Person>>() {
ProcessorContext context;
@Override
public void init(final ProcessorContext context) {
this.context = context;
}
@Override
public KeyValue<Integer, Person> transform(final PersonKey key, final Person value) {
// clear all headers; would be sufficient to only remove type header
for (Header h : context.headers().toArray()) {
context.headers().remove(h.key());
}
// effectively same as "old" selectKey code:
return KeyValue.pair(key.getId(), value);
}
@Override
public void close() {}
};
}}),
(school, person) -> {
System.out.println("school_app2= " + school + ", person_app2= " + person);
return null;
},
JoinWindows.of(Duration.ofSeconds(1)),
Joined.with(Serdes.Integer(), new SchoolSerde(), new PersonSerde())
);
}
}
interface KStreamProcessorX {
@Input("person")
KStream<?, ?> inputPersonKStream();
@Input("school")
KStream<?, ?> inputSchoolKStream();
@Input("school_1")
KStream<?, ?> inputSchool1KStream();
}
}