change to read from kafka
added schema
deserializer added