<commit_before>package com.continuuity.common.logging;

import com.continuuity.common.conf.CConfiguration;
import com.google.common.base.Objects;
import org.apache.avro.AvroRemoteException;
import org.apache.flume.source.AvroSource;
import org.apache.flume.source.avro.AvroFlumeEvent;
import org.apache.flume.source.avro.AvroSourceProtocol;
import org.apache.flume.source.avro.Status;
import org.apache.thrift.TBaseHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class FlumeLogAdapter implements AvroSourceProtocol {

  private static final Logger LOG
    = LoggerFactory.getLogger(FlumeLogAdapter.class);
  private CConfiguration config;
  private LogCollector collector;

  public FlumeLogAdapter(CConfiguration config) {
    this.config = config;
    this.collector = new LogCollector(config);
  }

  private static Map<String, String>
  toStringMap(Map<CharSequence, CharSequence> charSeqMap) {
    Map<String, String> stringMap =
      new HashMap<String, String>();
    if(charSeqMap == null) {
      return stringMap;
    }
    for(Map.Entry<CharSequence, CharSequence> entry : charSeqMap.entrySet()) {
      stringMap.put(entry.getKey().toString(), entry.getValue().toString());
    }
    return stringMap;
  }

  @Override
  public Status append(AvroFlumeEvent event) throws AvroRemoteException {

    Map<String, String> headers = toStringMap(event.getHeaders());
    if(! headers.containsKey(LogEvent.FIELD_NAME_LOGTAG) ||
       ! headers.containsKey(LogEvent.FIELD_NAME_LOGLEVEL)) {
      return Status.UNKNOWN;
    }

    String logtag = headers.get(LogEvent.FIELD_NAME_LOGTAG);
    String level = headers.get(LogEvent.FIELD_NAME_LOGLEVEL);

    if(logtag == null || logtag.isEmpty() || level == null || level.isEmpty()) {
      LOG.warn("Logtag or level is null. Please check the send events.");
      return Status.UNKNOWN;
    }

    String body = new String(TBaseHelper.byteBufferToByteArray(event.getBody()));
    LogEvent logEvent = new LogEvent(logtag, level, body);
    this.collector.log(logEvent);
    return Status.OK;
  }

  @Override
  public Status appendBatch(List<AvroFlumeEvent> events) throws AvroRemoteException {
    Status status = Status.OK;
    for (AvroFlumeEvent event : events) {
      Status stat = append(event);
      if (stat.equals(Status.FAILED) ||
          stat.equals(Status.UNKNOWN) && (!status.equals(Status.FAILED)))
        status = stat;
    }
    return status;
  }
}<commit_msg>Converted map of charsequence to char string for reading flume headers and also lowered log level<commit_after>package com.continuuity.common.logging;

import com.continuuity.common.conf.CConfiguration;
import com.google.common.base.Objects;
import org.apache.avro.AvroRemoteException;
import org.apache.flume.source.AvroSource;
import org.apache.flume.source.avro.AvroFlumeEvent;
import org.apache.flume.source.avro.AvroSourceProtocol;
import org.apache.flume.source.avro.Status;
import org.apache.thrift.TBaseHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class FlumeLogAdapter implements AvroSourceProtocol {

  private static final Logger LOG
    = LoggerFactory.getLogger(FlumeLogAdapter.class);
  private CConfiguration config;
  private LogCollector collector;

  public FlumeLogAdapter(CConfiguration config) {
    this.config = config;
    this.collector = new LogCollector(config);
  }

  private static Map<String, String>
  toStringMap(Map<CharSequence, CharSequence> charSeqMap) {
    Map<String, String> stringMap =
      new HashMap<String, String>();
    if(charSeqMap == null) {
      return stringMap;
    }
    for(Map.Entry<CharSequence, CharSequence> entry : charSeqMap.entrySet()) {
      stringMap.put(entry.getKey().toString(), entry.getValue().toString());
    }
    return stringMap;
  }

  @Override
  public Status append(AvroFlumeEvent event) throws AvroRemoteException {
    Map<String, String> headers = toStringMap(event.getHeaders());
    if(! headers.containsKey(LogEvent.FIELD_NAME_LOGTAG) ||
       ! headers.containsKey(LogEvent.FIELD_NAME_LOGLEVEL)) {
      return Status.UNKNOWN;
    }

    String logtag = headers.get(LogEvent.FIELD_NAME_LOGTAG);
    String level = headers.get(LogEvent.FIELD_NAME_LOGLEVEL);

    if(logtag == null || logtag.isEmpty() || level == null || level.isEmpty()) {
      LOG.warn("Logtag or level is null. Please check the send events.");
      return Status.UNKNOWN;
    }

    String body = new String(TBaseHelper.byteBufferToByteArray(event.getBody()));
    LogEvent logEvent = new LogEvent(logtag, level, body);
    this.collector.log(logEvent);
    return Status.OK;
  }

  @Override
  public Status appendBatch(List<AvroFlumeEvent> events) throws AvroRemoteException {
    Status status = Status.OK;
    for (AvroFlumeEvent event : events) {
      Status stat = append(event);
      if (stat.equals(Status.FAILED) ||
          stat.equals(Status.UNKNOWN) && (!status.equals(Status.FAILED)))
        status = stat;
    }
    return status;
  }
}

{'fraction_non_alphanumeric': 0.10998834045860863, 'fraction_numerical': 0.000777302759424796, 'mean_word_length': 4.136726546906187, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '417870', 'n_tokens_mistral': 1559, 'n_tokens_neox': 1560, 'n_words': 327}