Skip not trusted package in Kafka consumer

userAVky 0
Viewed 27

I used in my second project same Kafka topic name like in project one but but message has different package and in first project I got exception "The class 'com.example.proj2' is not in the trusted packages". I do not want this message in project one. Is there any chance to skip this message? I tried to catch SerializationException and commit but did not help. Message comes to consumer again and again.

Consumer config:

    Map<String, Object> configurations = new HashMap<>();
    configurations.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
    configurations.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "true");
    configurations.put(ConsumerConfig.GROUP_ID_CONFIG, groupId );
    configurations.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId );
    configurations.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    configurations.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    configurations.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
    configurations.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, "52428800");
    configurations.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "52428800");
    configurations.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "3600000");
    configurations.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
    configurations.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.proj1");
    configurations.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
    consumer = new KafkaConsumer<Object, Object>(configurations);
    consumer.subscribe(topic);

Consuming code:

    try {
        ConsumerRecords<Object, Object> records = consumer.poll(Duration.ofMillis(10000));
        if(records.count() > 0) {
            handleMessages(records);
            consumer.commitSync();
        }
    } catch(SerializationException e) {
        LOGGER.error("Invalid package", e);
        consumer.commitSync();
    } catch (Exception e) {...}
spring-kafka apache-kafka java
1 Answers

The DefaultJackson2JavaTypeMapper (used in that JsonDeserializer) does this:

            if (!isTrustedPackage(classId)) {
                throw new IllegalArgumentException("The class '" + classId
                        + "' is not in the trusted packages: "
                        + this.trustedPackages + ". "
                        + "If you believe this class is safe to deserialize, please provide its name. "
                        + "If the serialization is only done by a trusted source, you can also enable "
                        + "trust all (*).");

You may consider to use a ErrorHandlingDeserializer wrapper to handle that type of error: https://docs.spring.io/spring-kafka/reference/html/#error-handling-deserializer

Artem Bilan0
Related