CODEWITHSUNDEEP
javaapache-kafkaspring-kafkakafka-producer-apiserialization
Vaibhav Shelar
Vaibhav Shelar

Reputation: 75

Class for key.serializer could not be found

I am sending Array of JSON data to Kafka topic using Spring boot Application, but getting the following error:

error :org.apache.kafka.common.config.ConfigException: Invalid value 
org.apache.kafka.common.serialization.StringSerializer; for 
configuration key.serializer: Class 
org.apache.kafka.common.serialization.StringSerializer; could not be found.

I have tried to change serialization configuration to this:

props.put("key.serializer",  org.apache.kafka.common.serialization.StringSerializer;");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer;");

to

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer;");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer;");

Configuration file and Service File code:

@Configuration
public class KafkaProducerConfig {

@Bean

private static ProducerFactory<String, String> producerConfig() {

            Properties props = new Properties();
            props.put("bootstrap.servers", "localhost:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("linger.ms", 1);
            props.put("buffer.memory", 33554432);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer;");
            props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer;");

            // The following properties are used by LiKafkaProducerImpl
            props.put("large.message.enabled", "true");
            props.put("max.message.segment.bytes", 1000 * 1024);
            props.put("segment.serializer", DefaultSegmentSerializer.class.getName());

            props.put("auditor.class", LoggingAuditor.class.getName());

            return new DefaultKafkaProducerFactory(props);
        }
}

@Service
public class KafkaSender {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSender.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${kafka.topic.name}")
    private String topicName;

    public void sendData(List<Student> student) {

        System.out.println("Inside Student"+ student.toString());
        System.out.println("Inside Student"+ student);

        // TODO Auto-generated method stub
        Map<String, Object> headers = new HashMap<>();
        headers.put(KafkaHeaders.TOPIC, topicName);     

        System.out.println("\nStudent=  " + headers);

        // Construct a JSONObject from a Map.
        JSONObject HeaderObject = new JSONObject(headers);
        System.out.println("\nUsing new JSONObject() ==> " + HeaderObject);
        final String record = HeaderObject.toString();

        final int recordSize = record.length();

        kafkaTemplate.send(new GenericMessage<>(student, headers));

        LOGGER.info("Data - " + student + " sent to Kafka Topic - " + topicName);
    }
}

POST json:

[{ "studentId": "Q45678123", "firstName": "abc", "lastName": "xyz", "age": "12", "address": { "apartment": "apt 123", "street": "street Info", "state": "state", "city": "city", "postCode": "12345" } }, { "studentId": "Q45678123", "firstName": "abc", "lastName": "xyz", "age": "12", "address": { "apartment": "apt 123", "street": "street Info", "state": "state", "city": "city", "postCode": "12345" } } ]

Upvotes: 1

Views: 7874

Answers (1)

OneCricketeer
OneCricketeer

Reputation: 191681

You need to remove the semicolon from the end of the values

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

Or you could use the class.getName() method like you did for the segment serializer, which I would suggest is safer because then it guarantees that the serializer you want is available at compile time

Upvotes: 3

Related Questions