imaffe commented on a change in pull request #18406:
URL: https://github.com/apache/flink/pull/18406#discussion_r839759578
##########
File path:
flink-connectors/flink-connector-pulsar/src/main/java/org/apache/flink/connector/pulsar/source/reader/deserializer/PulsarSchemaWrapper.java
##########
@@ -41,22 +43,45 @@
/** The serializable pulsar schema, it wrap the schema with type class. */
private final PulsarSchema<T> pulsarSchema;
+ private boolean isSchemaEvolutionEnabled;
+
public PulsarSchemaWrapper(PulsarSchema<T> pulsarSchema) {
this.pulsarSchema = pulsarSchema;
}
@Override
- public void deserialize(Message<byte[]> message, Collector<T> out) throws
Exception {
- Schema<T> schema = this.pulsarSchema.getPulsarSchema();
- byte[] bytes = message.getData();
- T instance = schema.decode(bytes);
+ public void open(
+ DeserializationSchema.InitializationContext context,
SourceConfiguration configuration)
+ throws Exception {
+ this.isSchemaEvolutionEnabled =
configuration.isEnableSchemaEvolution();
+ }
- out.collect(instance);
+ @Override
+ public void deserialize(Message<?> message, Collector<T> out) throws
Exception {
Review comment:
Would you mind suggest an alternative ? I remember we discussed earlier
and this is the discussed design, maybe I misundertood ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
