<commit_before>package org.apache.flume.conf.file;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.flume.Channel;
import org.apache.flume.ChannelFactory;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
import org.apache.flume.Sink;
import org.apache.flume.SinkFactory;
import org.apache.flume.SinkRunner;
import org.apache.flume.Source;
import org.apache.flume.SourceFactory;
import org.apache.flume.SourceRunner;
import org.apache.flume.conf.Configurables;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.flume.node.ConfigurationProvider;
import org.apache.flume.node.nodemanager.NodeConfigurationAware;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

public class JsonFileConfigurationProvider implements ConfigurationProvider {

  private static final Logger logger = LoggerFactory
      .getLogger(JsonFileConfigurationProvider.class);

  private File file;
  private ChannelFactory channelFactory;
  private SourceFactory sourceFactory;
  private SinkFactory sinkFactory;
  private NodeConfigurationAware configurationAware;

  private LifecycleState lifecycleState;
  private ScheduledExecutorService executorService;
  private CounterGroup counterGroup;

  public JsonFileConfigurationProvider() {
    lifecycleState = LifecycleState.IDLE;
    counterGroup = new CounterGroup();
  }

  @Override
  public String toString() {
    return "{ file:" + file + " counterGroup:" + counterGroup + " }";
  }

  @Override
  public void start() {
    logger.info("JSON configuration provider starting");

    Preconditions.checkState(file != null,
        "The parameter file must not be null");

    executorService = Executors
        .newScheduledThreadPool(1,
            new ThreadFactoryBuilder().setNameFormat("conf-file-poller-%d")
                .build());

    FileWatcherRunnable fileWatcherRunnable = new FileWatcherRunnable();

    fileWatcherRunnable.file = file;
    fileWatcherRunnable.counterGroup = counterGroup;

    executorService.scheduleAtFixedRate(fileWatcherRunnable, 0, 30,
        TimeUnit.SECONDS);

    lifecycleState = LifecycleState.START;

    logger.debug("JSON configuration provider started");
  }

  @Override
  public void stop() {
    logger.info("JSON configuration provider stopping");

    executorService.shutdown();

    while (!executorService.isTerminated()) {
      try {
        logger.debug("Waiting for file watcher to terminate");
        executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        logger.debug("Interrupted while waiting for file watcher to terminate");
        Thread.currentThread().interrupt();
      }
    }

    lifecycleState = LifecycleState.STOP;

    logger.debug("JSON configuration provider stopped");
  }

  private void loadSources(SimpleNodeConfiguration conf,
      List<Map<String, Object>> defs) throws InstantiationException {

    logger.debug("Loading sources");

    for (Map<String, Object> sourceDef : defs) {
      logger.debug("source:{}", sourceDef);

      if (sourceDef.containsKey("type")) {
        Source source = sourceFactory.create((String) sourceDef.get("type"));
        Channel channel = conf.getChannels().get(sourceDef.get("channel"));

        Context context = new Context();
        context.setParameters(sourceDef);

        Configurables.configure(source, context);

        if (channel != null) {
          source.setChannel(channel);
        } else {
          logger.warn(
              "No channel named {} - source:{} is likely non-functional.",
              source, sourceDef.get("channel"));
        }

        conf.getSourceRunners().put((String) sourceDef.get("name"),
            SourceRunner.forSource(source));
      } else {
        throw new IllegalArgumentException("Illegal source definition:"
            + sourceDef + " - Missing type.");
      }
    }
  }

  private void loadSinks(SimpleNodeConfiguration conf,
      List<Map<String, Object>> defs) throws InstantiationException {

    logger.debug("Loading sinks");

    for (Map<String, Object> sinkDef : defs) {
      logger.debug("sink:{}", sinkDef);

      if (sinkDef.containsKey("type")) {
        Sink sink = sinkFactory.create((String) sinkDef.get("type"));
        Channel channel = conf.getChannels().get(sinkDef.get("channel"));

        Context context = new Context();
        context.setParameters(sinkDef);

        if (channel != null) {
          sink.setChannel(channel);
        } else {
          logger.warn(
              "No channel named {} - sink:{} is likely non-functional.", sink,
              sinkDef.get("channel"));
        }

        conf.getSinkRunners().put((String) sinkDef.get("name"),
            SinkRunner.forSink(sink));
      }
    }
  }

  private void loadChannels(SimpleNodeConfiguration conf,
      List<Map<String, Object>> defs) throws InstantiationException {

    logger.debug("Loading channels");

    for (Map<String, Object> channelDef : defs) {
      logger.debug("channel:{}", channelDef);

      if (channelDef.containsKey("type")) {
        Channel channel = channelFactory
            .create((String) channelDef.get("type"));

        Context context = new Context();
        context.setParameters(channelDef);

        conf.getChannels().put((String) channelDef.get("name"), channel);
      }
    }
  }

  private synchronized void load() {
    SimpleNodeConfiguration flumeConf = new SimpleNodeConfiguration();
    ObjectMapper mapper = new ObjectMapper();
<commit_msg>- Added missing calls to Configurables.configure(t, ctx) for sinks and channels during config loading.<commit_after>package org.apache.flume.conf.file;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.flume.Channel;
import org.apache.flume.ChannelFactory;
import org.apache.flume.Context;
import org.apache.flume.CounterGroup;
import org.apache.flume.Sink;
import org.apache.flume.SinkFactory;
import org.apache.flume.SinkRunner;
import org.apache.flume.Source;
import org.apache.flume.SourceFactory;
import org.apache.flume.SourceRunner;
import org.apache.flume.conf.Configurables;
import org.apache.flume.lifecycle.LifecycleState;
import org.apache.flume.node.ConfigurationProvider;
import org.apache.flume.node.nodemanager.NodeConfigurationAware;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;

public class JsonFileConfigurationProvider implements ConfigurationProvider {

  private static final Logger logger = LoggerFactory
      .getLogger(JsonFileConfigurationProvider.class);

  private File file;
  private ChannelFactory channelFactory;
  private SourceFactory sourceFactory;
  private SinkFactory sinkFactory;
  private NodeConfigurationAware configurationAware;

  private LifecycleState lifecycleState;
  private ScheduledExecutorService executorService;
  private CounterGroup counterGroup;

  public JsonFileConfigurationProvider() {
    lifecycleState = LifecycleState.IDLE;
    counterGroup = new CounterGroup();
  }

  @Override
  public String toString() {
    return "{ file:" + file + " counterGroup:" + counterGroup + " }";
  }

  @Override
  public void start() {
    logger.info("JSON configuration provider starting");

    Preconditions.checkState(file != null,
        "The parameter file must not be null");

    executorService = Executors
        .newScheduledThreadPool(1,
            new ThreadFactoryBuilder().setNameFormat("conf-file-poller-%d")
                .build());

    FileWatcherRunnable fileWatcherRunnable = new FileWatcherRunnable();

    fileWatcherRunnable.file = file;
    fileWatcherRunnable.counterGroup = counterGroup;

    executorService.scheduleAtFixedRate(fileWatcherRunnable, 0, 30,
        TimeUnit.SECONDS);

    lifecycleState = LifecycleState.START;

    logger.debug("JSON configuration provider started");
  }

  @Override
  public void stop() {
    logger.info("JSON configuration provider stopping");

    executorService.shutdown();

    while (!executorService.isTerminated()) {
      try {
        logger.debug("Waiting for file watcher to terminate");
        executorService.awaitTermination(500, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        logger.debug("Interrupted while waiting for file watcher to terminate");
        Thread.currentThread().interrupt();
      }
    }

    lifecycleState = LifecycleState.STOP;

    logger.debug("JSON configuration provider stopped");
  }

  private void loadSources(SimpleNodeConfiguration conf,
      List<Map<String, Object>> defs) throws InstantiationException {

    logger.debug("Loading sources");

    for (Map<String, Object> sourceDef : defs) {
      logger.debug("source:{}", sourceDef);

      if (sourceDef.containsKey("type")) {
        Source source = sourceFactory.create((String) sourceDef.get("type"));
        Channel channel = conf.getChannels().get(sourceDef.get("channel"));

        Context context = new Context();
        context.setParameters(sourceDef);

        Configurables.configure(source, context);

        if (channel != null) {
          source.setChannel(channel);
        } else {
          logger.warn(
              "No channel named {} - source:{} is likely non-functional.",
              source, sourceDef.get("channel"));
        }

        conf.getSourceRunners().put((String) sourceDef.get("name"),
            SourceRunner.forSource(source));
      } else {
        throw new IllegalArgumentException("Illegal source definition:"
            + sourceDef + " - Missing type.");
      }
    }
  }

  private void loadSinks(SimpleNodeConfiguration conf,
      List<Map<String, Object>> defs) throws InstantiationException {

    logger.debug("Loading sinks");

    for (Map<String, Object> sinkDef : defs) {
      logger.debug("sink:{}", sinkDef);

      if (sinkDef.containsKey("type")) {
        Sink sink = sinkFactory.create((String) sinkDef.get("type"));
        Channel channel = conf.getChannels().get(sinkDef.get("channel"));

        Context context = new Context();
        context.setParameters(sinkDef);

        Configurables.configure(sink, context);

        if (channel != null) {
          sink.setChannel(channel);
        } else {
          logger.warn(
              "No channel named {} - sink:{} is likely non-functional.", sink,
              sinkDef.get("channel"));
        }

        conf.getSinkRunners().put((String) sinkDef.get("name"),
            SinkRunner.forSink(sink));
      }
    }
  }

  private void loadChannels(SimpleNodeConfiguration conf,
      List<Map<String, Object>> defs) throws InstantiationException {

    logger.debug("Loading channels");

    for (Map<String, Object> channelDef : defs) {
      logger.debug("channel:{}", channelDef);

      if (channelDef.containsKey("type")) {
        Channel channel = channelFactory
            .create((String) channelDef.get("type"));

        Context context = new Context();
        context.setParameters(channelDef);

        Configurables.configure(channel, context);

        conf.getChannels().put((String) channelDef.get("name"), channel);
      }
    }
  }

  private synchronized void load() {
    SimpleNodeConfiguration flumeConf = new SimpleNodeConfiguration();
    ObjectMapper mapper = new ObjectMapper();

{'fraction_non_alphanumeric': 0.11079827757535608, 'fraction_numerical': 0.0014905597880092746, 'mean_word_length': 3.50802538260545, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, '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': '5400423', 'n_tokens_mistral': 3367, 'n_tokens_neox': 3245, 'n_words': 707}