// Path: src/main/java/com/google/cloud/solutions/datalineage/converter/MessageParser.java
// public interface MessageParser {
// 
//   /**
//    * Returns a POJO read at the provided path or null if not found.
//    */
//   @Nullable
//   <T> T read(String path);
// 
//   /** Returns a POJO read at the provided path or the provided default value if target is empty. */
//   <T> T readOrDefault(String path, T defaultValue);
// 
//   /** Returns true if the given key is present in the message. */
//   boolean containsKey(String keyPath);
// }


import static com.google.common.base.MoreObjects.firstNonNull;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.commons.lang3.StringUtils.isBlank;
import com.google.cloud.solutions.datalineage.converter.MessageParser;
import com.google.common.base.Objects;
import com.google.common.flogger.GoogleLogger;
import com.jayway.jsonpath.DocumentContext;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.PathNotFoundException;
import java.util.concurrent.TimeUnit;

  /**
   * Returns the JSON message used by this parser instant.
   */
  public String getJson() {
    return parsedMessage.jsonString();
  }

  public String getRootPath() {
    return rootPath;
  }

  @Override
  public <T> T read(String jsonPath) {
    try {
      return parsedMessage.read(buildPath(jsonPath));
    } catch (PathNotFoundException | NullPointerException exception) {
      logger.atInfo()
          .withCause(exception)
          .atMostEvery(1, TimeUnit.MINUTES)
          .log("error reading [%s]", jsonPath);
      return null;
    }
  }

  @Override
  public <T> T readOrDefault(String jsonPath, T defaultValue) {
    return firstNonNull(read(jsonPath), defaultValue);
  }

// Path: src/main/java/com/google/cloud/solutions/datalineage/converter/MessageParser.java
// public interface MessageParser {
// 
//   /**
//    * Returns a POJO read at the provided path or null if not found.
//    */
//   @Nullable
//   <T> T read(String path);
// 
//   /** Returns a POJO read at the provided path or the provided default value if target is empty. */
//   <T> T readOrDefault(String path, T defaultValue);
// 
//   /** Returns true if the given key is present in the message. */
//   boolean containsKey(String keyPath);
// }


// Path: src/main/java/com/google/cloud/solutions/datalineage/extractor/JsonMessageParser.java
import static com.google.common.base.MoreObjects.firstNonNull;
import static com.google.common.base.Preconditions.checkArgument;
import static org.apache.commons.lang3.StringUtils.isBlank;
import com.google.cloud.solutions.datalineage.converter.MessageParser;
import com.google.common.base.Objects;
import com.google.common.flogger.GoogleLogger;
import com.jayway.jsonpath.DocumentContext;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.PathNotFoundException;
import java.util.concurrent.TimeUnit;


  /**
   * Returns the JSON message used by this parser instant.
   */
  public String getJson() {
    return parsedMessage.jsonString();
  }

  public String getRootPath() {
    return rootPath;
  }

  @Override
  public <T> T read(String jsonPath) {
    try {
      return parsedMessage.read(buildPath(jsonPath));
    } catch (PathNotFoundException | NullPointerException exception) {
      logger.atInfo()
          .withCause(exception)
          .atMostEvery(1, TimeUnit.MINUTES)
          .log("error reading [%s]", jsonPath);
      return null;
    }
  }

  @Override
  public <T> T readOrDefault(String jsonPath, T defaultValue) {
    return firstNonNull(read(jsonPath), defaultValue);
  }

  @Override