title: What should be the best way to filter the kafka message
body: <p>I'm consuming data from a kafka topic which includes the area code. I have to filter the data only for certain area codes. Can any one suggest be the best approach to solve this.</p>
  
  <p>Here is my listener code looks like. Is it best practice to parse the data into object(as I mapped the payload to a TEST object) and filter the data based on the value which I need to filter or Does kafka provides any other libraries which I can make use of this filtering process. </p>
  
  <p><strong>Kafka Listener Method</strong></p>
  
  <pre><code>@Service
  public class Listener{
  
      @KafkaListener(topics = "#{@topicName}")
          public void listen(String payload) throws IOException {
  
              LOGGER.info("received payload from topic='{}'", payload);
              ObjectMapper objectMapper = new ObjectMapper();
              objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
  
              TEST test = objectMapper.readValue(payload,TEST.class);
  
          }
  }
  </code></pre>
  
  <p>My Kafka Configuration class:</p>
  
  <pre><code>@Configuration
  public class Config {
  
  
      @Bean
      public Map&lt;String, Object&gt; consumerConfigs() {
          Map&lt;String, Object&gt; properties = new HashMap&lt;&gt;();
          properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, applicationConfiguration.getKafkaBootStrap());
          properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, applicationConfiguration.getKafkaKeyDeserializer());
          properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, applicationConfiguration.getKafkaValueDeserializer());
          properties.put(ConsumerConfig.GROUP_ID_CONFIG, applicationConfiguration.getKafkaGroupId());
          properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, applicationConfiguration.getKafkaAutoOffsetReset());
          return properties;
      }
  
      @Bean
      public ConsumerFactory&lt;String, String&gt; consumerFactory() {
          return new DefaultKafkaConsumerFactory&lt;&gt;(consumerConfigs());
      }
  
      @Bean
      public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, String&gt;&gt; kafkaListenerContainerFactory() {
          ConcurrentKafkaListenerContainerFactory&lt;String, String&gt; factory =
                  new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          factory.setConsumerFactory(consumerFactory());
  
          return factory;
      }
      @Bean
      public Listener receiver() {
          return new Listener();
      }
  
  }
  </code></pre>