content: /**
   * DBSyncer Copyright 2020-2023 All Rights Reserved.
   */
  package org.dbsyncer.connector.kafka.serialization;
  
  import org.apache.kafka.common.errors.SerializationException;
  import org.apache.kafka.common.serialization.Deserializer;
  import org.dbsyncer.common.util.JsonUtil;
  
  import java.io.UnsupportedEncodingException;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  
  /**
   * @Author AE86
   * @Version 1.0.0
   * @Date 2021-12-16 23:09
   */
  public class JsonToMapDeserializer implements Deserializer<Map> {
      private String encoding = "UTF8";
  
      @Override
      public void configure(Map<String, ?> configs, boolean isKey) {
          String propertyName = isKey ? "key.deserializer.encoding" : "value.deserializer.encoding";
          Object encodingValue = configs.get(propertyName);
          if (encodingValue == null) {
              encodingValue = configs.get("deserializer.encoding");
          }
          if (encodingValue != null && encodingValue instanceof String) {
              encoding = (String) encodingValue;
          }
      }
  
      @Override
      public Map deserialize(String topic, byte[] data) {
          try {
              if (data == null) {
                  return null;
              }
              return JsonUtil.jsonToObj(new String(data, encoding), ConcurrentHashMap.class);
          } catch (UnsupportedEncodingException e) {
              throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + encoding);
          }
      }
  
      @Override
      public void close() {
          // nothing to do
      }
  }