import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
public ElasticSearchRestClient(String[] hostNames,
    ElasticSearchEventSerializer serializer) {

  for (int i = 0; i < hostNames.length; ++i) {
    if (!hostNames[i].contains("http://") && !hostNames[i].contains("https://")) {
      hostNames[i] = "http://" + hostNames[i];
    }
  }
  this.serializer = serializer;

  serversList = new RoundRobinList<String>(Arrays.asList(hostNames));
  httpClient = new DefaultHttpClient();
  bulkBuilder = new StringBuilder();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
/**
 *
 * @param clientType
 *    String representation of client type
 * @param hostNames
 *    Array of strings that represents hostnames with ports (hostname:port)
 * @param clusterName
 *    Elasticsearch cluster name used only by Transport Client
 * @param serializer
 *    Serializer of flume events to elasticsearch documents
 * @return
 */
public ElasticSearchClient getClient(String clientType, String[] hostNames,
    String clusterName, ElasticSearchEventSerializer serializer,
    ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
  if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
    return new ElasticSearchTransportClient(hostNames, clusterName, serializer);
  } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) { 
    return new ElasticSearchTransportClient(hostNames, clusterName, indexBuilder);
  } else if (clientType.equalsIgnoreCase(RestClient) && serializer != null) {
    return new ElasticSearchRestClient(hostNames, serializer);
  }
  throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
/**
 * Used for tests only. Creates local elasticsearch instance client.
 *
 * @param clientType Name of client to use
 * @param serializer Serializer for the event
 * @param indexBuilder Index builder factory
 *
 * @return Local elastic search instance client
 */
public ElasticSearchClient getLocalClient(String clientType,
                                          ElasticSearchEventSerializer serializer,
                                          ElasticSearchIndexRequestBuilderFactory indexBuilder)
    throws NoSuchClientTypeException {
  if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
    return new ElasticSearchTransportClient(serializer);
  } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null)  {
    return new ElasticSearchTransportClient(indexBuilder);
  } else if (clientType.equalsIgnoreCase(RestClient)) {
  }
  throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
/**
 *
 * @param clientType String representation of client type
 * @param hostNames Array of strings that represents hosntames with ports
 * (hostname:port)
 * @param clusterName Elasticsearch cluster name used only by Transport
 * Client
 * @param serializer Serializer of flume events to elasticsearch documents
 * @return
 */
public ElasticSearchClient getClient(String clientType, String[] hostNames,
        String clusterName, ElasticSearchEventSerializer serializer,
        ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
    if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
        return new ElasticSearchTransportClient(hostNames, clusterName, serializer);
    } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
        return new ElasticSearchTransportClient(hostNames, clusterName, indexBuilder);
    } else if (clientType.equalsIgnoreCase(RestClient) && serializer != null) {
        return new ElasticSearchRestClient(hostNames, serializer);
    }
    throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
/**
 * Used for tests only. Creates local elasticsearch instance client.
 *
 * @param clientType Name of client to use
 * @param serializer Serializer for the event
 * @param indexBuilder Index builder factory
 *
 * @return Local elastic search instance client
 */
public ElasticSearchClient getLocalClient(String clientType, ElasticSearchEventSerializer serializer,
        ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
    if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
        return new ElasticSearchTransportClient(serializer);
    } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
        return new ElasticSearchTransportClient(indexBuilder);
    } else if (clientType.equalsIgnoreCase(RestClient)) {
    }
    throw new NoSuchClientTypeException();
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
public ExtendedElasticSearchIndexRequestBuilderFactory(ElasticSearchEventSerializer serializer) {
	super(FastDateFormat.getInstance("yyyy.MM.dd", TimeZone.getTimeZone("Etc/UTC")));

	this.serializer = serializer;
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
public LogstashEventSerializerIndexRequestBuilderFactory(ElasticSearchEventSerializer serializer) {
    super(serializer);
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
/**
 * Transport client for external cluster
 * 
 * @param hostNames
 * @param clusterName
 * @param serializer
 */
public ElasticSearchTransportClient(String[] hostNames, String clusterName,
    ElasticSearchEventSerializer serializer) {
  configureHostnames(hostNames);
  this.serializer = serializer;
  openClient(clusterName);
}
 

import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer; //导入依赖的package包/类
/**
 * Transport client for external cluster
 *
 * @param hostNames
 * @param clusterName
 * @param serializer
 */
public ElasticSearchTransportClient(String[] hostNames, String clusterName,
        ElasticSearchEventSerializer serializer) {
    configureHostnames(hostNames);
    this.serializer = serializer;
    openClient(clusterName);
}