code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.source.mysql;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import io.airbyte.db.jdbc.JdbcDatabase;
  import io.airbyte.integrations.debezium.CdcTargetPosition;
  import io.airbyte.integrations.debezium.internals.SnapshotMetadata;
  import java.sql.SQLException;
  import java.util.List;
  import java.util.Objects;
  import java.util.stream.Stream;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class MySqlCdcTargetPosition implements CdcTargetPosition {
  
    private static final Logger LOGGER = LoggerFactory.getLogger(MySqlCdcTargetPosition.class);
    public final String fileName;
    public final Integer position;
  
    public MySqlCdcTargetPosition(final String fileName, final Integer position) {
      this.fileName = fileName;
      this.position = position;
    }
  
    @Override
    public boolean equals(final Object obj) {
      if (obj instanceof final MySqlCdcTargetPosition cdcTargetPosition) {
        return fileName.equals(cdcTargetPosition.fileName) && cdcTargetPosition.position.equals(position);
      }
      return false;
    }
  
    @Override
    public int hashCode() {
      return Objects.hash(fileName, position);
    }
  
    @Override
    public String toString() {
      return "FileName: " + fileName + ", Position : " + position;
    }
  
    public static MySqlCdcTargetPosition targetPosition(final JdbcDatabase database) {
      try (final Stream<MySqlCdcTargetPosition> stream = database.unsafeResultSetQuery(
          connection -> connection.createStatement().executeQuery("SHOW MASTER STATUS"),
          resultSet -> {
            final String file = resultSet.getString("File");
            final int position = resultSet.getInt("Position");
            if (file == null || position == 0) {
              return new MySqlCdcTargetPosition(null, null);
            }
            return new MySqlCdcTargetPosition(file, position);
          })) {
        final List<MySqlCdcTargetPosition> masterStatus = stream.toList();
        final MySqlCdcTargetPosition targetPosition = masterStatus.get(0);
        LOGGER.info("Target File position : " + targetPosition);
        return targetPosition;
      } catch (final SQLException e) {
        throw new RuntimeException(e);
      }
  
    }
  
    @Override
    public boolean reachedTargetPosition(final JsonNode valueAsJson) {
      final String eventFileName = valueAsJson.get("source").get("file").asText();
      final int eventPosition = valueAsJson.get("source").get("pos").asInt();
  
      final boolean isSnapshot = SnapshotMetadata.TRUE == SnapshotMetadata.valueOf(
          valueAsJson.get("source").get("snapshot").asText().toUpperCase());
  
      if (isSnapshot || fileName.compareTo(eventFileName) > 0
          || (fileName.compareTo(eventFileName) == 0 && position >= eventPosition)) {
        return false;
      }
  
      LOGGER.info("Signalling close because record's binlog file : " + eventFileName + " , position : " + eventPosition
          + " is after target file : "
          + fileName + " , target position : " + position);
      return true;
    }
  
  }
  