code: @Override
    public void onPartitionProcessBegin(Partition partition, long partitionProcessTime, long partitionUpdateTime) {
  
      Preconditions.checkNotNull(partition);
      Preconditions.checkNotNull(partition.getTable());
  
      if (!this.expectedHighWatermarks.hasPartitionWatermarks(tableKey(partition.getTable()))) {
        throw new IllegalStateException(String.format(
            "onPartitionProcessBegin called before onTableProcessBegin for table: %s, partitions: %s",
            tableKey(partition.getTable()), partitionKey(partition)));
      }
  
      // Remove dropped partitions
      Collection<String> droppedPartitions =
          Collections2.transform(AbstractAvroToOrcConverter.getDropPartitionsDDLInfo(partition),
              new Function<Map<String, String>, String>() {
                @Override
                public String apply(Map<String, String> input) {
                  return PARTITION_VALUES_JOINER.join(input.values());
                }
              });
  
      this.expectedHighWatermarks.removePartitionWatermarks(tableKey(partition.getTable()), droppedPartitions);
      this.expectedHighWatermarks.addPartitionWatermark(tableKey(partition.getTable()), partitionKey(partition),
          partitionUpdateTime);
    }
docs: Adds an expected high watermark for this {@link Partition}. Also removes any watermarks for partitions being replaced.
  Replace partitions are read using partition parameter {@link AbstractAvroToOrcConverter#REPLACED_PARTITIONS_HIVE_METASTORE_KEY}.
  Uses the <code>partitionUpdateTime</code> as the high watermark for this <code>partition</code>
  
  {@inheritDoc}
  @see org.apache.gobblin.data.management.conversion.hive.watermarker.HiveSourceWatermarker#onPartitionProcessBegin(org.apache.hadoop.hive.ql.metadata.Partition, long, long)