<commit_before>package org.gbif.occurrence.cli.registry.sync;

import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.api.model.registry.Dataset;
import org.gbif.api.model.registry.Organization;
import org.gbif.api.service.registry.DatasetService;
import org.gbif.api.service.registry.OrganizationService;
import org.gbif.common.messaging.api.MessagePublisher;
import org.gbif.common.messaging.api.messages.DeleteDatasetOccurrencesMessage;
import org.gbif.common.messaging.api.messages.OccurrenceDeletionReason;
import org.gbif.common.messaging.api.messages.OccurrenceMutatedMessage;
import org.gbif.common.messaging.guice.PostalServiceModule;
import org.gbif.occurrence.cli.registry.RegistryObjectMapperContextResolver;
import org.gbif.occurrence.common.config.OccHBaseConfiguration;
import org.gbif.occurrence.persistence.api.OccurrencePersistenceService;
import org.gbif.occurrence.persistence.guice.OccurrencePersistenceModule;
import org.gbif.registry.ws.client.DatasetWsClient;
import org.gbif.registry.ws.client.OrganizationWsClient;
import org.gbif.ws.mixin.Mixins;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.sun.jersey.api.client.Client;
import com.sun.jersey.api.client.WebResource;
import com.sun.jersey.api.client.config.ClientConfig;
import com.sun.jersey.api.client.config.DefaultClientConfig;
import com.sun.jersey.client.apache.ApacheHttpClient;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.codehaus.jackson.jaxrs.JacksonJsonProvider;
import org.codehaus.jackson.map.DeserializationConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A mapreduce Mapper that synchronizes occurrences with the registry. It checks for changes detected by
 * {@link RegistryBasedOccurrenceMutator} and dataset deletions. For organization changes the new values are written
 * to the occurrence HBase table via occurrence persistence, and then an OccurrenceMutatedMessage is sent. For dataset
 * deletions a DeleteDatasetMessage is sent.
 */
public class OccurrenceScanMapper extends TableMapper<ImmutableBytesWritable, NullWritable> {

  private static final Logger LOG = LoggerFactory.getLogger(OccurrenceScanMapper.class);

  private static final Set<UUID> DEAD_DATASETS = Sets.newHashSet();
  private static final Set<UUID> UNCHANGED_DATASETS = Sets.newHashSet();
  private static final Map<UUID, Organization> DATASET_TO_OWNING_ORG = Maps.newHashMap();

  private DatasetService datasetService;
  private OrganizationService orgService;
  private RegistryBasedOccurrenceMutator occurrenceMutator;
  private OccurrencePersistenceService occurrencePersistenceService;
  private MessagePublisher messagePublisher;
  private Client httpClient;

  private int numRecords = 0;

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    super.setup(context);
    Properties props = new Properties();
    // extract the config properties from the job context
    for (Map.Entry<String, String> entry : context.getConfiguration()) {
      props.setProperty(entry.getKey(), entry.getValue());
    }

    ClientConfig cc = new DefaultClientConfig();
    cc.getClasses().add(JacksonJsonProvider.class);
    cc.getClasses().add(RegistryObjectMapperContextResolver.class);
    cc.getFeatures().put(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES.toString(), false);
    RegistryObjectMapperContextResolver.addMixIns(Mixins.getPredefinedMixins());

    httpClient = ApacheHttpClient.create(cc);
    WebResource regResource = httpClient.resource(props.getProperty(SyncCommon.REG_WS_PROPS_KEY));
    datasetService = new DatasetWsClient(regResource, null);
    orgService = new OrganizationWsClient(regResource, null);
    occurrenceMutator = new RegistryBasedOccurrenceMutator();

    OccHBaseConfiguration occHBaseConfiguration = new OccHBaseConfiguration();
    occHBaseConfiguration.occTable = props.getProperty("occurrence.db.table_name");
    occHBaseConfiguration.zkConnectionString = props.getProperty("occurrence.db.zookeeper.connection_string");

    Injector injector =
      Guice.createInjector(new PostalServiceModule("sync", props),
              new OccurrencePersistenceModule(occHBaseConfiguration, context.getConfiguration()));
    occurrencePersistenceService = injector.getInstance(OccurrencePersistenceService.class);
    messagePublisher = injector.getInstance(MessagePublisher.class);
  }

  @Override
  protected void cleanup(Context context) throws IOException, InterruptedException {
    if (messagePublisher != null) {
      messagePublisher.close();
    }
    if (httpClient != null) {
      httpClient.destroy();
    }
  }

  @Override
  public void map(ImmutableBytesWritable row, Result values, Context context) throws IOException {
    UUID datasetKey = UUID.fromString(Bytes.toString(values.getValue(SyncCommon.OCC_CF, SyncCommon.DK_COL)));
    if (DEAD_DATASETS.contains(datasetKey) || UNCHANGED_DATASETS.contains(datasetKey)) {
      return;
    }

    Dataset dataset = datasetService.get(datasetKey);
    if (dataset.getDeleted() != null) {
      DEAD_DATASETS.add(datasetKey);
      try {
        LOG.info("Sending delete dataset message for dataset [{}]", datasetKey);
        messagePublisher.send(new DeleteDatasetOccurrencesMessage(datasetKey, OccurrenceDeletionReason.DATASET_MANUAL));
      } catch (IOException e) {
        LOG.warn("Failed to send update message", e);<commit_msg>ensure httpClient is always closed  even if messagePublisher throws an exception<commit_after>package org.gbif.occurrence.cli.registry.sync;

import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.api.model.registry.Dataset;
import org.gbif.api.model.registry.Organization;
import org.gbif.api.service.registry.DatasetService;
import org.gbif.api.service.registry.OrganizationService;
import org.gbif.common.messaging.api.MessagePublisher;
import org.gbif.common.messaging.api.messages.DeleteDatasetOccurrencesMessage;
import org.gbif.common.messaging.api.messages.OccurrenceDeletionReason;
import org.gbif.common.messaging.api.messages.OccurrenceMutatedMessage;
import org.gbif.common.messaging.guice.PostalServiceModule;
import org.gbif.occurrence.cli.registry.RegistryObjectMapperContextResolver;
import org.gbif.occurrence.common.config.OccHBaseConfiguration;
import org.gbif.occurrence.persistence.api.OccurrencePersistenceService;
import org.gbif.occurrence.persistence.guice.OccurrencePersistenceModule;
import org.gbif.registry.ws.client.DatasetWsClient;
import org.gbif.registry.ws.client.OrganizationWsClient;
import org.gbif.ws.mixin.Mixins;

import java.io.IOException;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.sun.jersey.api.client.Client;
import com.sun.jersey.api.client.WebResource;
import com.sun.jersey.api.client.config.ClientConfig;
import com.sun.jersey.api.client.config.DefaultClientConfig;
import com.sun.jersey.client.apache.ApacheHttpClient;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import org.codehaus.jackson.jaxrs.JacksonJsonProvider;
import org.codehaus.jackson.map.DeserializationConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A mapreduce Mapper that synchronizes occurrences with the registry. It checks for changes detected by
 * {@link RegistryBasedOccurrenceMutator} and dataset deletions. For organization changes the new values are written
 * to the occurrence HBase table via occurrence persistence, and then an OccurrenceMutatedMessage is sent. For dataset
 * deletions a DeleteDatasetMessage is sent.
 */
public class OccurrenceScanMapper extends TableMapper<ImmutableBytesWritable, NullWritable> {

  private static final Logger LOG = LoggerFactory.getLogger(OccurrenceScanMapper.class);

  private static final Set<UUID> DEAD_DATASETS = Sets.newHashSet();
  private static final Set<UUID> UNCHANGED_DATASETS = Sets.newHashSet();
  private static final Map<UUID, Organization> DATASET_TO_OWNING_ORG = Maps.newHashMap();

  private DatasetService datasetService;
  private OrganizationService orgService;
  private RegistryBasedOccurrenceMutator occurrenceMutator;
  private OccurrencePersistenceService occurrencePersistenceService;
  private MessagePublisher messagePublisher;
  private Client httpClient;

  private int numRecords = 0;

  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
    super.setup(context);
    Properties props = new Properties();
    // extract the config properties from the job context
    for (Map.Entry<String, String> entry : context.getConfiguration()) {
      props.setProperty(entry.getKey(), entry.getValue());
    }

    ClientConfig cc = new DefaultClientConfig();
    cc.getClasses().add(JacksonJsonProvider.class);
    cc.getClasses().add(RegistryObjectMapperContextResolver.class);
    cc.getFeatures().put(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES.toString(), false);
    RegistryObjectMapperContextResolver.addMixIns(Mixins.getPredefinedMixins());

    httpClient = ApacheHttpClient.create(cc);
    WebResource regResource = httpClient.resource(props.getProperty(SyncCommon.REG_WS_PROPS_KEY));
    datasetService = new DatasetWsClient(regResource, null);
    orgService = new OrganizationWsClient(regResource, null);
    occurrenceMutator = new RegistryBasedOccurrenceMutator();

    OccHBaseConfiguration occHBaseConfiguration = new OccHBaseConfiguration();
    occHBaseConfiguration.occTable = props.getProperty("occurrence.db.table_name");
    occHBaseConfiguration.zkConnectionString = props.getProperty("occurrence.db.zookeeper.connection_string");

    Injector injector =
      Guice.createInjector(new PostalServiceModule("sync", props),
              new OccurrencePersistenceModule(occHBaseConfiguration, context.getConfiguration()));
    occurrencePersistenceService = injector.getInstance(OccurrencePersistenceService.class);
    messagePublisher = injector.getInstance(MessagePublisher.class);
  }

  @Override
  protected void cleanup(Context context) throws IOException, InterruptedException {
    if (messagePublisher != null) {
      // wrapped to ensure we also close the httpClient if messagePublisher throws an exception (including runtime)
      try {
        messagePublisher.close();
      }
      catch (Exception ignore) {}
    }

    if (httpClient != null) {
      httpClient.destroy();
    }
  }

  @Override
  public void map(ImmutableBytesWritable row, Result values, Context context) throws IOException {
    UUID datasetKey = UUID.fromString(Bytes.toString(values.getValue(SyncCommon.OCC_CF, SyncCommon.DK_COL)));
    if (DEAD_DATASETS.contains(datasetKey) || UNCHANGED_DATASETS.contains(datasetKey)) {
      return;
    }

    Dataset dataset = datasetService.get(datasetKey);
    if (dataset.getDeleted() != null) {
      DEAD_DATASETS.add(datasetKey);
      try {
        LOG.info("Sending delete dataset message for dataset [{}]", datasetKey);
        messagePublisher.send(new DeleteDatasetOccurrencesMessage(datasetKey, OccurrenceDeletionReason.DATASET_MANUAL));
      } catch (IOException e) {
        LOG.warn("Failed to send update message", e);
{'fraction_non_alphanumeric': 0.09280080550427924, 'fraction_numerical': 0.0005034401745259272, 'mean_word_length': 5.534539473684211, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '265149', 'n_tokens_mistral': 3460, 'n_tokens_neox': 3358, 'n_words': 666}