package com.cloudera.cdk.flume;

import com.cloudera.data.Dataset;
import com.cloudera.data.DatasetRepository;
import com.cloudera.data.FieldPartitioner;
import com.cloudera.data.PartitionKey;
import com.cloudera.data.PartitionStrategy;
import java.lang.reflect.Constructor;
import java.net.URI;
import java.net.URL;
import java.util.Map;
import org.apache.avro.Schema;
import org.apache.flume.FlumeException;

public class Log4jAppender extends org.apache.flume.clients.log4jappender.Log4jAppender {

  private static final String PARTITION_PREFIX = "cdk.partition.";

  private String datasetRepositoryClass;
  private String datasetRepositoryUri;
  private String datasetName;
  
  private PartitionStrategy partitionStrategy;
  private PartitionKey key;

  public Log4jAppender() {
    super();
    setAvroReflectionEnabled(true);
  }

  /**
   * Sets the hostname and port. Even if these are passed the
   * <tt>activateOptions()</tt> function must be called before calling
   * <tt>append()</tt>, else <tt>append()</tt> will throw an Exception.
   * @param hostname The first hop where the client should connect to.
   * @param port The port to connect on the host.
   *
   */
  public Log4jAppender(String hostname, int port) {
    super(hostname, port);
    setAvroReflectionEnabled(true);
  }

  public void setDatasetRepositoryClass(String datasetRepositoryClass) {
    this.datasetRepositoryClass = datasetRepositoryClass;
  }

  public void setDatasetRepositoryUri(String datasetRepositoryUri) {
    this.datasetRepositoryUri = datasetRepositoryUri;
  }

  public void setDatasetName(String datasetName) {
    this.datasetName = datasetName;
  }

  @Override
  public void activateOptions() throws FlumeException {
    try {
      Class<?> c = Class.forName(datasetRepositoryClass);
      Constructor<?> cons = c.getConstructor(URI.class);
      DatasetRepository repo = (DatasetRepository)
          cons.newInstance(new URI(datasetRepositoryUri));

      Dataset dataset = repo.get(datasetName);
      if (dataset.getDescriptor().isPartitioned()) {
        partitionStrategy = dataset.getDescriptor().getPartitionStrategy();
      }
      URL schemaUrl = dataset.getDescriptor().getSchemaUrl();
      if (schemaUrl != null) {
        setAvroSchemaUrl(schemaUrl.toExternalForm());
      }
    } catch (Exception e) {
      throw new FlumeException(e);
    }

    super.activateOptions();
  }

  protected void populateAvroHeaders(Map<String, String> hdrs, Schema schema,
      Object message) {
    super.populateAvroHeaders(hdrs, schema, message);
    if (partitionStrategy != null) {
      key = partitionStrategy.partitionKeyForEntity(message, key);
      int i = 0;
      for (FieldPartitioner fp : partitionStrategy.getFieldPartitioners()) {
        hdrs.put(PARTITION_PREFIX + fp.getName(), key.get(i++).toString());
      }
    }
  }
}

@@ -673,16 +673,47 @@
 etName;%0A
+  private boolean initialized;%0A
   %0A  pri
@@ -1702,71 +1702,267 @@
 %0A  p
-ublic void activateOptions() throws FlumeException %7B%0A
+rotected void populateAvroHeaders(Map%3CString, String%3E hdrs, Schema schema,%0A      Object message) %7B%0A    if (!initialized) %7B%0A      // initialize here rather than in activateOptions to avoid initialization%0A      // cycle in Configuration and log4j%0A 
     
+ 
 try %7B%0A
+  
     
@@ -2021,16 +2021,18 @@
 ;%0A      
+  
 Construc
@@ -2076,24 +2076,26 @@
 ass);%0A      
+ 
+ 
 DatasetRepos
@@ -2127,16 +2127,18 @@
 sitory)%0A
+  
         
@@ -2191,24 +2191,26 @@
 i));%0A%0A      
+  
 Dataset data
@@ -2240,24 +2240,26 @@
 ame);%0A      
+ 
+ 
 if (dataset.
@@ -2289,24 +2289,26 @@
 tioned()) %7B%0A
+  
         part
@@ -2381,16 +2381,18 @@
     
+  
 %7D%0A      
 URL 
@@ -2387,16 +2387,18 @@
 %7D%0A      
+ 
+ 
 URL sche
@@ -2447,24 +2447,26 @@
 rl();%0A      
+  
 if (schemaUr
@@ -2482,24 +2482,26 @@
 ) %7B%0A        
+ 
+ 
 setAvroSchem
@@ -2530,24 +2530,26 @@
 nalForm());%0A
+  
       %7D%0A    
@@ -2548,16 +2548,18 @@
   %7D%0A    
+  
 %7D catch 
@@ -2570,24 +2570,26 @@
 eption e) %7B%0A
+  
       throw 
@@ -2619,146 +2619,63 @@
     
-%7D%0A%0A    super.activateOptions();%0A  %7D%0A%0A  protected void populateAvroHeaders(Map%3CString, String%3E hdrs, Schema schema,%0A      Object message) %7B
+  %7D finally %7B%0A        initialized = true;%0A      %7D%0A    %7D
 %0A   

Fix initialization ordering issue in Log4jAppender. Found while working on CDK-39.