Custom Kafka Deserializer и Spring’овый контекст. Как инжектить в статические поля

В блоге beeline cloud я рассказывал о Spring Data JPA, Hibernate, делился личными наблюдениями, как облегчить себе жизнь при написании тестов. Сегодня речь о другом: расскажу, как инжектить в статические поля. Как всегда — на примерах. Поехали.

На практике десериализаторов, представленных в ядре Spring, хватает в 99% случаев. Но бывают ситуации, когда всё же требуется описать свою логику предобработки входящего сообщения.

Для наглядности предлагаю рассмотреть гипотетический случай, когда на вход Kafka подается строка лога с задаваемым разделителем. В этой строке нас интересует лишь часть сообщения: мы будем делить строку по определенному символу и собирать из получившегося массива новую строку.

Задавать разделитель и индексы в файле application.properties мы планируем так:

kafka.message.deserializer.delimiter=, kafka.message.deserializer.indexes=0,2,3,5

Сразу скажу, что org.apache.kafka.common.serialization.Deserializer и все его реализации не знают о спринговом контексте ничего — от слова совсем. Даже если попытаться сделать из него бин, пометив соответствующей аннотацией, например @Component.

 Для этого напишем сам Deserializer и попробуем прямо в него заинжектить требуемые настройки, используя спринговую аннотацию @Value:

@Slf4j @Component public class SmartStringDeserializer implements Deserializer<String> {    @Value("${kafka.message.deserializer.delimiter}")    String delimiter;      @Value("${kafka.message.deserializer.indexes}")    List<Integer> indexes;    @Override    public void configure(Map<String, ?> configs, boolean isKey) {        Deserializer.super.configure(configs, isKey);    }      @Override    public String deserialize(String s, byte[] bytes) {        return setValueByIndex(bytes);    }      @Override    public void close() {        Deserializer.super.close();    }      private String setValueByIndex(byte[] bytes) {        String msg = new String(bytes, StandardCharsets.UTF_8);        LOGGER.debug("Income message: {}", msg);        if (indexes.getFirst() == -1) {        return msg;        }        String[] incomeMessage = msg.split(delimiter);        String formattedIncomeMessage = indexes.stream()            .map(index -> incomeMessage[index])            .collect(Collectors.joining(delimiter));        LOGGER.debug("Consumed message: {}", formattedIncomeMessage);          return formattedIncomeMessage;    } }

К сожалению, нам вернется null при попытке получить значения:

А если мы проставим стандартные значения в @Values, результат будет таким же:

 @Value(«${kafka.message.deserializer.delimiter:,}»)

 Но можно использовать статические поля, описав их в соседнем классе. Вот только надо придумать, как в них заинжектить данные из спрингового контекста. Конечно, это должен быть спринговый бин, например @Component, и в нем мы определим статические поля, включая поля, которые будут заполняться данными из properties:

@Component public class DeserializerUtils {      public static String DELIMITER;      public static List<Integer> INDEXES;      private String delimiter;      private List<Integer> indexes;    … }

Теперь напишем сеттеры, они-то и будут присваивать значения статическим полям:

@Component public class DeserializerUtils {      public static String DELIMITER;      public static List<Integer> INDEXES;      private String delimiter;      private List<Integer> indexes;      @Value("${kafka.message.deserializer.delimiter:,}")    private void setDelimiter(String delimiter){        DeserializerUtils.DELIMITER = delimiter;    }      @Value("${kafka.message.deserializer.indexes:-1}")    private void setIndexes(List<Integer> indexes){        DeserializerUtils.INDEXES = indexes;    } }

Важно помнить, что @Values необходимо навешивать над методами, иначе не заработает. И, если вы решите обновить контекст без перезапуска приложения, например, поправив значения в файле конфигураций, обновятся и статические поля.

 Теперь перепишем сам десериалайзер:

@Slf4j public class SmartStringDeserializer implements Deserializer<String> {      @Override    public void configure(Map<String, ?> configs, boolean isKey) {        Deserializer.super.configure(configs, isKey);    }      @Override    public String deserialize(String s, byte[] bytes) {        return setValueByIndex(bytes);    }      @Override    public void close() {        Deserializer.super.close();    }      private String setValueByIndex(byte[] bytes) {        String delimiter = DeserializerUtils.DELIMITER;        List<Integer> indexes = DeserializerUtils.INDEXES;        String msg = new String(bytes, StandardCharsets.UTF_8);        LOGGER.debug("Income message: {}", msg);        if (indexes.getFirst() == -1) {        return msg;        }        String[] incomeMessage = msg.split(delimiter);        String formattedIncomeMessage = indexes.stream()            .map(index -> incomeMessage[index])            .collect(Collectors.joining(delimiter));        LOGGER.debug("Consumed message: {}", formattedIncomeMessage);          return formattedIncomeMessage;    } }

А вот теперь мы можем конфигурировать Deserializer, используя спринговый контекст:

У такого подхода есть свои минусы. По факту у нас дублируются данные в хипе, которые не удалит GC ввиду того, что первые хранятся в контексте спринга, вторые — в статических полях. Поэтому старайтесь инжектить конкретные бины и не делайте так:

public static ApplicationContext CONTEXT; private ApplicationContext context;   private void setIndexes(ApplicationContext context){    DeserializerUtils.CONTEXT = context; }

beeline cloud — secure cloud provider. Разрабатываем облачные решения, чтобы вы предоставляли клиентам лучшие сервисы.