Avro schema not found for topic X

[irajhedayati]

Related to #456 but different scenario

Version: 0.27

Command to run:
sudo docker run  \
    -e ZOOKEEPER_QUORUM='ZK1:2181,ZK2:2181,ZK3:2181' \
    -e ZOOKEEPER_PATH='/' \
    -e AWS_SECRET_KEY='*****' \
    -e AWS_ACCESS_KEY='*****' \
    -e SECOR_GROUP='secor_avro' \
    -e SECOR_S3_BUCKET='*******' \
    -e SECOR_MESSAGE_TIMESTAMP='recordDateTime' \
    -e SECOR_EXTRA_OPTS='-Dsecor.compression.codec=org.apache.hadoop.io.compress.GzipCodec -Dsecor.file.reader.writer.factory=com.pinterest.secor.io.impl.AvroParquetFileReaderWriterFactory -Dkafka.useTimestamp=true' \
    -e SECOR_KAFKA_TOPIC_FILTER='^avro_topics.*$' \
    -e SECOR_SCHEMA_REGISTRY=http://sr:8081 \
    -e SECOR_MESSAGE_PARSER='com.pinterest.secor.parser.AvroMessageParser' \
    -e S3_PATH='secor_parquet' \
    -e SECOR_MAX_FILE_SECONDS='300' \
    -e SECOR_MAX_FILE_BYTES='200000' \
    docker.repo.local/secor:0.27

The topic avro_topics_x in Kafka has Avro messages. As usual, the schema registered under the subject of avro_topics_x-value. Once the application started, it will throw the following exception when writing the messages:

Caused by: java.lang.IllegalStateException: Avro schema not found for topic avro_topics_x
        at com.pinterest.secor.common.SecorSchemaRegistryClient.getSchema(SecorSchemaRegistryClient.java:50)
        at com.pinterest.secor.io.impl.AvroParquetFileReaderWriterFactory$AvroParquetFileWriter.<init>(AvroParquetFileReaderWriterFactory.java:113)
        at com.pinterest.secor.io.impl.AvroParquetFileReaderWriterFactory.BuildFileWriter(AvroParquetFileReaderWriterFactory.java:56)
        at com.pinterest.secor.util.ReflectionUtil.createFileWriter(ReflectionUtil.java:154)
        at com.pinterest.secor.common.FileRegistry.getOrCreateWriter(FileRegistry.java:133)
        at com.pinterest.secor.writer.MessageWriter.write(MessageWriter.java:94)
        at com.pinterest.secor.consumer.Consumer.consumeNextMessage(Consumer.java:194)
        ... 1 more

As discussed #456 the issue is related to the time and place of parsing the message. If kafka.useTimestamp=true, it won't parse the message till gets to the writer.

• Parsing message is done by com.pinterest.secor.common.SecorSchemaRegistryClient#decodeMessage that goes to KafkaAvroDeserializer and uses the proper way of getting the schema
• Getting a schema for a topic uses com.pinterest.secor.common.SecorSchemaRegistryClient#getSchema that just queries the internal cache (a map). The schema won't be there unless you had a call to decodeMessage before.

[eugenemiretsky]

Did you manage to solve this? We are facing the same issue.
Seems like a very basic use case.

[eugenemiretsky]

kafka.useTimestamp=false did the trick

[irajhedayati]

I used the configuration. At the very least, the documentation should be updated when using either of

• com.pinterest.secor.io.impl.AvroFileReaderWriterFactory
• com.pinterest.secor.io.impl.AvroParquetFileReaderWriterFactory
  you should set kafka.useTimestamp=false.

On another note, it could be a design matter. In order to set the schema of the Avro/Parquet file, it uses the topic name (schemaRegistry.getSchema(topic)) and one would say we could just call Schema Registry to get the latest. The current implementation is also using a map that keeps the latest seen in the messages. However, it would be a problem in case of schema evolution and processing historical data. Say on day X, we have the schema v1 and then we update it to v2 on day Y. Then if you run the code, it will try to save data of all days including X and before with v2 which is wrong.

I'm gonna create a PR to get the schema by topic and then try to use the ID as well.

[irajhedayati]

The fix is merged but forgot to close it.