Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Avoid ClassCastException for the same class when used with Flink #509

Closed
wants to merge 1 commit into from
Closed

Avoid ClassCastException for the same class when used with Flink #509

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
9 changes: 8 additions & 1 deletion ...erializer/src/main/java/io/confluent/kafka/serializers/AbstractKafkaAvroDeserializer.java
View file
Open in desktop
Expand Up @@ -43,6 +43,7 @@ public abstract class AbstractKafkaAvroDeserializer extends AbstractKafkaAvroSer

private final DecoderFactory decoderFactory = DecoderFactory.get();
protected boolean useSpecificAvroReader = false;
protected boolean forceNewSpecificDataInstance = false;
private final Map<String, Schema> readerSchemaCache = new ConcurrentHashMap<String, Schema>();


Expand All @@ -54,6 +55,8 @@ protected void configure(KafkaAvroDeserializerConfig config) {
configureClientProperties(config);
useSpecificAvroReader = config
.getBoolean(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG);
forceNewSpecificDataInstance = config
.getBoolean(KafkaAvroDeserializerConfig.AVRO_FORCE_NEW_SPECIFIC_DATA_CONFIG);
}

protected KafkaAvroDeserializerConfig deserializerConfig(Map<String, ?> props) {
Expand Down Expand Up @@ -201,7 +204,11 @@ private DatumReader getDatumReader(Schema writerSchema, Schema readerSchema) {
if (readerSchema == null) {
readerSchema = getReaderSchema(writerSchema);
}
return new SpecificDatumReader(writerSchema, readerSchema);
if (useSpecificAvroReader && forceNewSpecificDataInstance) {
return new SpecificDatumReader(writerSchema, readerSchema, new SpecificData());
} else {
return new SpecificDatumReader(writerSchema, readerSchema);
}
} else {
if (readerSchema == null) {
return new GenericDatumReader(writerSchema);
Expand Down
8 changes: 7 additions & 1 deletion ...-serializer/src/main/java/io/confluent/kafka/serializers/KafkaAvroDeserializerConfig.java
View file
Open in desktop
Expand Up @@ -28,13 +28,19 @@ public class KafkaAvroDeserializerConfig extends AbstractKafkaAvroSerDeConfig {
public static final boolean SPECIFIC_AVRO_READER_DEFAULT = false;
public static final String SPECIFIC_AVRO_READER_DOC =
"If true, tries to look up the SpecificRecord class ";

public static final String AVRO_FORCE_NEW_SPECIFIC_DATA_CONFIG = "force.new.specific.avro.instance";
public static final boolean AVRO_FORCE_NEW_SPECIFIC_DATA_DEFAULT = false;
public static final String AVRO_FORCE_NEW_SPECIFIC_DATA_DOC = "If true, it passes a new instace of SpecificData to SpecificDatumReader ";
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good to see that this is at Importance.LOW, but the docs could probably be clearer. This is a very specific scenario where you'd need to use this setting. We should probably call out exactly when you'd want to use it.


private static ConfigDef config;

static {
config = baseConfigDef()
.define(SPECIFIC_AVRO_READER_CONFIG, Type.BOOLEAN, SPECIFIC_AVRO_READER_DEFAULT,
Importance.LOW, SPECIFIC_AVRO_READER_DOC);
Importance.LOW, SPECIFIC_AVRO_READER_DOC)
.define(AVRO_FORCE_NEW_SPECIFIC_DATA_CONFIG, Type.BOOLEAN, AVRO_FORCE_NEW_SPECIFIC_DATA_DEFAULT,
Importance.LOW, AVRO_FORCE_NEW_SPECIFIC_DATA_DOC);
}

public KafkaAvroDeserializerConfig(Map<?, ?> props) {
Expand Down