<commit_before>/*
 * Copyright 2008-2009 LinkedIn, Inc
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package voldemort.client;

import java.io.StringReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import voldemort.client.protocol.RequestFormatType;
import voldemort.cluster.Cluster;
import voldemort.cluster.Node;
import voldemort.cluster.failuredetector.FailureDetector;
import voldemort.serialization.ByteArraySerializer;
import voldemort.serialization.IdentitySerializer;
import voldemort.serialization.SerializationException;
import voldemort.serialization.Serializer;
import voldemort.serialization.SerializerDefinition;
import voldemort.serialization.SerializerFactory;
import voldemort.serialization.SlopSerializer;
import voldemort.serialization.StringSerializer;
import voldemort.store.Store;
import voldemort.store.StoreDefinition;
import voldemort.store.compress.CompressingStore;
import voldemort.store.compress.CompressionStrategy;
import voldemort.store.compress.CompressionStrategyFactory;
import voldemort.store.logging.LoggingStore;
import voldemort.store.metadata.MetadataStore;
import voldemort.store.nonblockingstore.NonblockingStore;
import voldemort.store.routed.RoutedStoreFactory;
import voldemort.store.serialized.SerializingStore;
import voldemort.store.slop.Slop;
import voldemort.store.stats.StatTrackingStore;
import voldemort.store.stats.StoreStats;
import voldemort.store.stats.StoreStatsJmx;
import voldemort.store.versioned.InconsistencyResolvingStore;
import voldemort.utils.ByteArray;
import voldemort.utils.JmxUtils;
import voldemort.versioning.ChainedResolver;
import voldemort.versioning.InconsistencyResolver;
import voldemort.versioning.TimeBasedInconsistencyResolver;
import voldemort.versioning.VectorClockInconsistencyResolver;
import voldemort.versioning.Versioned;
import voldemort.xml.ClusterMapper;
import voldemort.xml.StoreDefinitionsMapper;

import com.google.common.collect.Maps;

/**
 * A base class for various {@link voldemort.client.StoreClientFactory
 * StoreClientFactory} implementations
 * 
 * 
 */
public abstract class AbstractStoreClientFactory implements StoreClientFactory {

    private static AtomicInteger jmxIdCounter = new AtomicInteger(0);

    public static final int DEFAULT_ROUTING_TIMEOUT_MS = 5000;

    protected static final ClusterMapper clusterMapper = new ClusterMapper();
    private static final StoreDefinitionsMapper storeMapper = new StoreDefinitionsMapper();
    protected static final Logger logger = Logger.getLogger(AbstractStoreClientFactory.class);

    private static final Serializer<ByteArray> slopKeySerializer = new ByteArraySerializer();
    private static final Serializer<Slop> slopValueSerializer = new SlopSerializer();

    private final URI[] bootstrapUrls;
    private final ExecutorService threadPool;
    private final SerializerFactory serializerFactory;
    private final boolean isJmxEnabled;
    private final RequestFormatType requestFormatType;
    private final int jmxId;
    protected volatile FailureDetector failureDetector;
    private final int maxBootstrapRetries;
    private final StoreStats stats;
    private final ClientConfig config;
    private final RoutedStoreFactory routedStoreFactory;
    private final int clientZoneId;

    public AbstractStoreClientFactory(ClientConfig config) {
        this.config = config;
        this.threadPool = new ClientThreadPool(config.getMaxThreads(),
                                               config.getThreadIdleTime(TimeUnit.MILLISECONDS),
                                               config.getMaxQueuedRequests());
        this.serializerFactory = config.getSerializerFactory();
        this.bootstrapUrls = validateUrls(config.getBootstrapUrls());
        this.isJmxEnabled = config.isJmxEnabled();
        this.requestFormatType = config.getRequestFormatType();
        this.jmxId = jmxIdCounter.getAndIncrement();
        this.maxBootstrapRetries = config.getMaxBootstrapRetries();
        this.stats = new StoreStats();
        this.clientZoneId = config.getClientZoneId();
        this.routedStoreFactory = new RoutedStoreFactory(config.isPipelineRoutedStoreEnabled(),
                                                         threadPool,
                                                         config.getRoutingTimeout(TimeUnit.MILLISECONDS));

        if(this.isJmxEnabled) {
            JmxUtils.registerMbean(threadPool,
                                   JmxUtils.createObjectName(JmxUtils.getPackageName(threadPool.getClass()),
                                                             JmxUtils.getClassName(threadPool.getClass())
                                                                     + jmxId()));
            JmxUtils.registerMbean(new StoreStatsJmx(stats),
                                   JmxUtils.createObjectName("voldemort.store.stats.aggregate",
                                                             "aggregate-perf" + jmxId()));
        }
    }

    public <K, V> StoreClient<K, V> getStoreClient(String storeName) {
        return getStoreClient(storeName, null);
    }

    public <K, V> StoreClient<K, V> getStoreClient(String storeName,
                                                   InconsistencyResolver<Versioned<V>> resolver) {
        return new DefaultStoreClient<K, V>(storeName, resolver, this, 3);
    }

    @SuppressWarnings("unchecked")
    public <K, V, T> Store<K, V, T> getRawStore(String storeName,
                                                InconsistencyResolver<Versioned<V>> resolver) {
        // Get cluster and store metadata
        String clusterXml = bootstrapMetadataWithRetries(MetadataStore.CLUSTER_KEY, bootstrapUrls);
        Cluster cluster = clusterMapper.readCluster(new StringReader(clusterXml), false);
        String storesXml = bootstrapMetadataWithRetries(MetadataStore.STORES_KEY, bootstrapUrls);
        List<StoreDefinition> storeDefs = storeMapper.readStoreList(new StringReader(storesXml),
                                                                    false);
        StoreDefinition storeDef = null;
        for(StoreDefinition d: storeDefs)
            if(d.getName().equals(storeName))
                storeDef = d;
        if(storeDef == null)
            throw new BootstrapFailureException("Unknown store '" + storeName + "'.");

        boolean repairReads = !storeDef.isView();

        // construct mapping
        Map<Integer, Store<ByteArray, byte[], byte[]>> clientMapping = Maps.newHashMap();
        Map<Integer, NonblockingStore> nonblockingStores = Maps.newHashMap();
        Map<Integer, NonblockingStore> nonblockingSlopStores = Maps.newHashMap();

        Map<Integer, Store<ByteArray, Slop, byte[]>> slopStores = null;
        if(storeDef.hasHintedHandoffStrategyType())
            slopStores = Maps.newHashMap();

        for(Node node: cluster.getNodes()) {
            Store<ByteArray, byte[], byte[]> store = getStore(storeDef.getName(),
                                                              node.getHost(),
                                                              getPort(node),
                                                              this.requestFormatType);
            Store<ByteArray, byte[], byte[]> loggingStore = new LoggingStore(store);
            clientMapping.put(node.getId(), loggingStore);

            NonblockingStore nonblockingStore = routedStoreFactory.toNonblockingStore(store);
            nonblockingStores.put(node.getId(), nonblockingStore);

            if(slopStores != null) {
                Store<ByteArray, byte[], byte[]> rawSlopStore = getStore("slop",
                                                                         node.getHost(),
                                                                         getPort(node),
                                                                         this.requestFormatType);
                Store<ByteArray, Slop, byte[]> slopStore = SerializingStore.wrap(rawSlopStore,
                                                                                 slopKeySerializer,
                                                                                 slopValueSerializer,
                                                                                 new IdentitySerializer());
                slopStores.put(node.getId(), slopStore);
                nonblockingSlopStores.put(node.getId(), routedStoreFactory.toNonblockingStore(rawSlopStore));
            }
        }

        Store<ByteArray, byte[], byte[]> store = routedStoreFactory.create(cluster,
                                                                           storeDef,
                                                                           clientMapping,
                                                                           nonblockingStores,
                                                                           slopStores,
                                                                           nonblockingSlopStores,
                                                                           repairReads,
                                                                           clientZoneId,
                                                                           getFailureDetector());

        if(isJmxEnabled) {
            StatTrackingStore statStore = new StatTrackingStore(store, this.stats);
            store = statStore;
            JmxUtils.registerMbean(new StoreStatsJmx(statStore.getStats()),
                                   JmxUtils.createObjectName(JmxUtils.getPackageName(store.getClass()),
                                                             store.getName() + jmxId()));
        }

        if(storeDef.getKeySerializer().hasCompression()
           || storeDef.getValueSerializer().hasCompression()) {
            store = new CompressingStore(store,
                                         getCompressionStrategy(storeDef.getKeySerializer()),
                                         getCompressionStrategy(storeDef.getValueSerializer()));
        }

        Serializer<K> keySerializer = (Serializer<K>) serializerFactory.getSerializer(storeDef.getKeySerializer());
        Serializer<V> valueSerializer = (Serializer<V>) serializerFactory.getSerializer(storeDef.getValueSerializer());
<commit_msg>add logging for voldemort client calls<commit_after>/*
 * Copyright 2008-2009 LinkedIn, Inc
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package voldemort.client;

import java.io.StringReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import voldemort.client.protocol.RequestFormatType;
import voldemort.cluster.Cluster;
import voldemort.cluster.Node;
import voldemort.cluster.failuredetector.FailureDetector;
import voldemort.serialization.ByteArraySerializer;
import voldemort.serialization.IdentitySerializer;
import voldemort.serialization.SerializationException;
import voldemort.serialization.Serializer;
import voldemort.serialization.SerializerDefinition;
import voldemort.serialization.SerializerFactory;
import voldemort.serialization.SlopSerializer;
import voldemort.serialization.StringSerializer;
import voldemort.store.Store;
import voldemort.store.StoreDefinition;
import voldemort.store.compress.CompressingStore;
import voldemort.store.compress.CompressionStrategy;
import voldemort.store.compress.CompressionStrategyFactory;
import voldemort.store.logging.LoggingStore;
import voldemort.store.metadata.MetadataStore;
import voldemort.store.nonblockingstore.NonblockingStore;
import voldemort.store.routed.RoutedStoreFactory;
import voldemort.store.serialized.SerializingStore;
import voldemort.store.slop.Slop;
import voldemort.store.stats.StatTrackingStore;
import voldemort.store.stats.StoreStats;
import voldemort.store.stats.StoreStatsJmx;
import voldemort.store.versioned.InconsistencyResolvingStore;
import voldemort.utils.ByteArray;
import voldemort.utils.JmxUtils;
import voldemort.versioning.ChainedResolver;
import voldemort.versioning.InconsistencyResolver;
import voldemort.versioning.TimeBasedInconsistencyResolver;
import voldemort.versioning.VectorClockInconsistencyResolver;
import voldemort.versioning.Versioned;
import voldemort.xml.ClusterMapper;
import voldemort.xml.StoreDefinitionsMapper;

import com.google.common.collect.Maps;

/**
 * A base class for various {@link voldemort.client.StoreClientFactory
 * StoreClientFactory} implementations
 * 
 * 
 */
public abstract class AbstractStoreClientFactory implements StoreClientFactory {

    private static AtomicInteger jmxIdCounter = new AtomicInteger(0);

    public static final int DEFAULT_ROUTING_TIMEOUT_MS = 5000;

    protected static final ClusterMapper clusterMapper = new ClusterMapper();
    private static final StoreDefinitionsMapper storeMapper = new StoreDefinitionsMapper();
    protected static final Logger logger = Logger.getLogger(AbstractStoreClientFactory.class);

    private static final Serializer<ByteArray> slopKeySerializer = new ByteArraySerializer();
    private static final Serializer<Slop> slopValueSerializer = new SlopSerializer();

    private final URI[] bootstrapUrls;
    private final ExecutorService threadPool;
    private final SerializerFactory serializerFactory;
    private final boolean isJmxEnabled;
    private final RequestFormatType requestFormatType;
    private final int jmxId;
    protected volatile FailureDetector failureDetector;
    private final int maxBootstrapRetries;
    private final StoreStats stats;
    private final ClientConfig config;
    private final RoutedStoreFactory routedStoreFactory;
    private final int clientZoneId;

    public AbstractStoreClientFactory(ClientConfig config) {
        this.config = config;
        this.threadPool = new ClientThreadPool(config.getMaxThreads(),
                                               config.getThreadIdleTime(TimeUnit.MILLISECONDS),
                                               config.getMaxQueuedRequests());
        this.serializerFactory = config.getSerializerFactory();
        this.bootstrapUrls = validateUrls(config.getBootstrapUrls());
        this.isJmxEnabled = config.isJmxEnabled();
        this.requestFormatType = config.getRequestFormatType();
        this.jmxId = jmxIdCounter.getAndIncrement();
        this.maxBootstrapRetries = config.getMaxBootstrapRetries();
        this.stats = new StoreStats();
        this.clientZoneId = config.getClientZoneId();
        this.routedStoreFactory = new RoutedStoreFactory(config.isPipelineRoutedStoreEnabled(),
                                                         threadPool,
                                                         config.getRoutingTimeout(TimeUnit.MILLISECONDS));

        if(this.isJmxEnabled) {
            JmxUtils.registerMbean(threadPool,
                                   JmxUtils.createObjectName(JmxUtils.getPackageName(threadPool.getClass()),
                                                             JmxUtils.getClassName(threadPool.getClass())
                                                                     + jmxId()));
            JmxUtils.registerMbean(new StoreStatsJmx(stats),
                                   JmxUtils.createObjectName("voldemort.store.stats.aggregate",
                                                             "aggregate-perf" + jmxId()));
        }
    }

    public <K, V> StoreClient<K, V> getStoreClient(String storeName) {
        return getStoreClient(storeName, null);
    }

    public <K, V> StoreClient<K, V> getStoreClient(String storeName,
                                                   InconsistencyResolver<Versioned<V>> resolver) {
        return new DefaultStoreClient<K, V>(storeName, resolver, this, 3);
    }

    @SuppressWarnings("unchecked")
    public <K, V, T> Store<K, V, T> getRawStore(String storeName,
                                                InconsistencyResolver<Versioned<V>> resolver) {
        // Get cluster and store metadata
        String clusterXml = bootstrapMetadataWithRetries(MetadataStore.CLUSTER_KEY, bootstrapUrls);
        Cluster cluster = clusterMapper.readCluster(new StringReader(clusterXml), false);
        String storesXml = bootstrapMetadataWithRetries(MetadataStore.STORES_KEY, bootstrapUrls);
        List<StoreDefinition> storeDefs = storeMapper.readStoreList(new StringReader(storesXml),
                                                                    false);
        StoreDefinition storeDef = null;
        for(StoreDefinition d: storeDefs)
            if(d.getName().equals(storeName))
                storeDef = d;
        if(storeDef == null)
            throw new BootstrapFailureException("Unknown store '" + storeName + "'.");

        boolean repairReads = !storeDef.isView();

        // construct mapping
        Map<Integer, Store<ByteArray, byte[], byte[]>> clientMapping = Maps.newHashMap();
        Map<Integer, NonblockingStore> nonblockingStores = Maps.newHashMap();
        Map<Integer, NonblockingStore> nonblockingSlopStores = Maps.newHashMap();

        Map<Integer, Store<ByteArray, Slop, byte[]>> slopStores = null;
        if(storeDef.hasHintedHandoffStrategyType())
            slopStores = Maps.newHashMap();

        for(Node node: cluster.getNodes()) {
            Store<ByteArray, byte[], byte[]> store = getStore(storeDef.getName(),
                                                              node.getHost(),
                                                              getPort(node),
                                                              this.requestFormatType);
            clientMapping.put(node.getId(), store);

            NonblockingStore nonblockingStore = routedStoreFactory.toNonblockingStore(store);
            nonblockingStores.put(node.getId(), nonblockingStore);

            if(slopStores != null) {
                Store<ByteArray, byte[], byte[]> rawSlopStore = getStore("slop",
                                                                         node.getHost(),
                                                                         getPort(node),
                                                                         this.requestFormatType);
                Store<ByteArray, Slop, byte[]> slopStore = SerializingStore.wrap(rawSlopStore,
                                                                                 slopKeySerializer,
                                                                                 slopValueSerializer,
                                                                                 new IdentitySerializer());
                slopStores.put(node.getId(), slopStore);
                nonblockingSlopStores.put(node.getId(),
                                          routedStoreFactory.toNonblockingStore(rawSlopStore));
            }
        }

        Store<ByteArray, byte[], byte[]> store = routedStoreFactory.create(cluster,
                                                                           storeDef,
                                                                           clientMapping,
                                                                           nonblockingStores,
                                                                           slopStores,
                                                                           nonblockingSlopStores,
                                                                           repairReads,
                                                                           clientZoneId,
                                                                           getFailureDetector());
        store = new LoggingStore(store);

        if(isJmxEnabled) {
            StatTrackingStore statStore = new StatTrackingStore(store, this.stats);
            store = statStore;
            JmxUtils.registerMbean(new StoreStatsJmx(statStore.getStats()),
                                   JmxUtils.createObjectName(JmxUtils.getPackageName(store.getClass()),
                                                             store.getName() + jmxId()));
        }

        if(storeDef.getKeySerializer().hasCompression()
           || storeDef.getValueSerializer().hasCompression()) {
            store = new CompressingStore(store,
                                         getCompressionStrategy(storeDef.getKeySerializer()),
                                         getCompressionStrategy(storeDef.getValueSerializer()));
        }

        Serializer<K> keySerializer = (Serializer<K>) serializerFactory.getSerializer(storeDef.getKeySerializer());
        Serializer<V> valueSerializer = (Serializer<V>) serializerFactory.getSerializer(storeDef.getValueSerializer());

{'fraction_non_alphanumeric': 0.07560909826485661, 'fraction_numerical': 0.0017081722556864156, 'mean_word_length': 2.0247450713800137, 'pattern_counts': {'":': 0, '<': 60, '<?xml version=': 0, '>': 60, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 4}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 158, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4227259', 'n_tokens_mistral': 5180, 'n_tokens_neox': 5000, 'n_words': 1048}