content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you 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 org.apache.apex.malhar.flume.sink;
  
  import java.io.IOError;
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.ServiceConfigurationError;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.flume.discovery.Discovery;
  import org.apache.apex.malhar.flume.sink.Server.Client;
  import org.apache.apex.malhar.flume.sink.Server.Request;
  import org.apache.apex.malhar.flume.storage.EventCodec;
  import org.apache.apex.malhar.flume.storage.Storage;
  
  import org.apache.flume.Context;
  import org.apache.flume.Event;
  import org.apache.flume.EventDeliveryException;
  import org.apache.flume.Transaction;
  import org.apache.flume.conf.Configurable;
  import org.apache.flume.sink.AbstractSink;
  
  import com.datatorrent.api.Component;
  import com.datatorrent.api.StreamCodec;
  
  import com.datatorrent.netlet.DefaultEventLoop;
  import com.datatorrent.netlet.NetletThrowable;
  import com.datatorrent.netlet.NetletThrowable.NetletRuntimeException;
  import com.datatorrent.netlet.util.Slice;
  
  /**
   * FlumeSink is a flume sink developed to ingest the data into DataTorrent DAG
   * from flume. It's essentially a flume sink which acts as a server capable of
   * talking to one client at a time. The client for this server is AbstractFlumeInputOperator.
   * <p />
   * &lt;experimental&gt;FlumeSink auto adjusts the rate at which it consumes the data from channel to
   * match the throughput of the DAG.&lt;/experimental&gt;
   * <p />
   * The properties you can set on the FlumeSink are: <br />
   * id - string unique value identifying this sink <br />
   * hostname - string value indicating the fqdn or ip address of the interface on which the server should listen <br />
   * port - integer value indicating the numeric port to which the server should bind <br />
   * sleepMillis - integer value indicating the number of milliseconds the process should sleep when there are no events
   * before checking for next event again <br />
   * throughputAdjustmentPercent - integer value indicating by what percentage the flume transaction size should be
   * adjusted upward or downward at a time <br />
   * minimumEventsPerTransaction - integer value indicating the minimum number of events per transaction <br />
   * maximumEventsPerTransaction - integer value indicating the maximum number of events per transaction. This value can
   * not be more than channel's transaction capacity.<br />
   *
   * @since 0.9.2
   */
  public class FlumeSink extends AbstractSink implements Configurable
  {
    private static final String HOSTNAME_STRING = "hostname";
    private static final String HOSTNAME_DEFAULT = "locahost";
    private static final long ACCEPTED_TOLERANCE = 20000;
    private DefaultEventLoop eventloop;
    private Server server;
    private int outstandingEventsCount;
    private int lastConsumedEventsCount;
    private int idleCount;
    private byte[] playback;
    private Client client;
    private String hostname;
    private int port;
    private String id;
    private long acceptedTolerance;
    private long sleepMillis;
    private double throughputAdjustmentFactor;
    private int minimumEventsPerTransaction;
    private int maximumEventsPerTransaction;
    private long commitEventTimeoutMillis;
    private transient long lastCommitEventTimeMillis;
    private Storage storage;
    Discovery<byte[]> discovery;
    StreamCodec<Event> codec;
    /* Begin implementing Flume Sink interface */
  
    @Override
    @SuppressWarnings({"BroadCatchBlock", "TooBroadCatch", "UseSpecificCatch", "SleepWhileInLoop"})
    public Status process() throws EventDeliveryException
    {
      Slice slice;
      synchronized (server.requests) {
        for (Request r : server.requests) {
          logger.debug("found {}", r);
          switch (r.type) {
            case SEEK:
              lastCommitEventTimeMillis = System.currentTimeMillis();
              slice = r.getAddress();
              playback = storage.retrieve(Arrays.copyOfRange(slice.buffer, slice.offset, slice.offset + slice.length));
              client = r.client;
              break;
  
            case COMMITTED:
              lastCommitEventTimeMillis = System.currentTimeMillis();
              slice = r.getAddress();
              storage.clean(Arrays.copyOfRange(slice.buffer, slice.offset, slice.offset + slice.length));
              break;
  
            case CONNECTED:
              logger.debug("Connected received, ignoring it!");
              break;
  
            case DISCONNECTED:
              if (r.client == client) {
                client = null;
                outstandingEventsCount = 0;
              }
              break;
  
            case WINDOWED:
              lastConsumedEventsCount = r.getEventCount();
              idleCount = r.getIdleCount();
              outstandingEventsCount -= lastConsumedEventsCount;
              break;
  
            case SERVER_ERROR:
              throw new IOError(null);
  
            default:
              logger.debug("Cannot understand the request {}", r);
              break;
          }
        }
  
        server.requests.clear();
      }
  
      if (client == null) {
        logger.info("No client expressed interest yet to consume the events.");
        return Status.BACKOFF;
      } else if (System.currentTimeMillis() - lastCommitEventTimeMillis > commitEventTimeoutMillis) {
        logger.info("Client has not processed the workload given for the last {} milliseconds, so backing off.",
            System.currentTimeMillis() - lastCommitEventTimeMillis);
        return Status.BACKOFF;
      }
  
      int maxTuples;
      // the following logic needs to be fixed... this is a quick put together.
      if (outstandingEventsCount < 0) {
        if (idleCount > 1) {
          maxTuples = (int)((1 + throughputAdjustmentFactor * idleCount) * lastConsumedEventsCount);
        } else {
          maxTuples = (int)((1 + throughputAdjustmentFactor) * lastConsumedEventsCount);
        }
      } else if (outstandingEventsCount > lastConsumedEventsCount) {
        maxTuples = (int)((1 - throughputAdjustmentFactor) * lastConsumedEventsCount);
      } else {
        if (idleCount > 0) {
          maxTuples = (int)((1 + throughputAdjustmentFactor * idleCount) * lastConsumedEventsCount);
          if (maxTuples <= 0) {
            maxTuples = minimumEventsPerTransaction;
          }
        } else {
          maxTuples = lastConsumedEventsCount;
        }
      }
  
      if (maxTuples >= maximumEventsPerTransaction) {
        maxTuples = maximumEventsPerTransaction;
      } else if (maxTuples <= 0) {
        maxTuples = minimumEventsPerTransaction;
      }
  
      if (maxTuples > 0) {
        if (playback != null) {
          try {
            int i = 0;
            do {
              if (!client.write(playback)) {
                retryWrite(playback, null);
              }
              outstandingEventsCount++;
              playback = storage.retrieveNext();
            }
            while (++i < maxTuples && playback != null);
          } catch (Exception ex) {
            logger.warn("Playback Failed", ex);
            if (ex instanceof NetletThrowable) {
              try {
                eventloop.disconnect(client);
              } finally {
                client = null;
                outstandingEventsCount = 0;
              }
            }
            return Status.BACKOFF;
          }
        } else {
          int storedTuples = 0;
  
          Transaction t = getChannel().getTransaction();
          try {
            t.begin();
  
            Event e;
            while (storedTuples < maxTuples && (e = getChannel().take()) != null) {
              Slice event = codec.toByteArray(e);
              byte[] address = storage.store(event);
              if (address != null) {
                if (!client.write(address, event)) {
                  retryWrite(address, event);
                }
                outstandingEventsCount++;
              } else {
                logger.debug("Detected the condition of recovery from flume crash!");
              }
              storedTuples++;
            }
  
            if (storedTuples > 0) {
              storage.flush();
            }
  
            t.commit();
  
            if (storedTuples > 0) { /* log less frequently */
              logger.debug("Transaction details maxTuples = {}, storedTuples = {}, outstanding = {}",
                  maxTuples, storedTuples, outstandingEventsCount);
            }
          } catch (Error er) {
            t.rollback();
            throw er;
          } catch (Exception ex) {
            logger.error("Transaction Failed", ex);
            if (ex instanceof NetletRuntimeException && client != null) {
              try {
                eventloop.disconnect(client);
              } finally {
                client = null;
                outstandingEventsCount = 0;
              }
            }
            t.rollback();
            return Status.BACKOFF;
          } finally {
            t.close();
          }
  
          if (storedTuples == 0) {
            sleep();
          }
        }
      }
  
      return Status.READY;
    }
  
    private void sleep()
    {
      try {
        Thread.sleep(sleepMillis);
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
      }
    }
  
    @Override
    public void start()
    {
      try {
        if (storage instanceof Component) {
          @SuppressWarnings("unchecked")
          Component<com.datatorrent.api.Context> component = (Component<com.datatorrent.api.Context>)storage;
          component.setup(null);
        }
        if (discovery instanceof Component) {
          @SuppressWarnings("unchecked")
          Component<com.datatorrent.api.Context> component = (Component<com.datatorrent.api.Context>)discovery;
          component.setup(null);
        }
        if (codec instanceof Component) {
          @SuppressWarnings("unchecked")
          Component<com.datatorrent.api.Context> component = (Component<com.datatorrent.api.Context>)codec;
          component.setup(null);
        }
        eventloop = new DefaultEventLoop("EventLoop-" + id);
        server = new Server(id, discovery,acceptedTolerance);
      } catch (Error error) {
        throw error;
      } catch (RuntimeException re) {
        throw re;
      } catch (IOException ex) {
        throw new RuntimeException(ex);
      }
  
      eventloop.start();
      eventloop.start(hostname, port, server);
      super.start();
    }
  
    @Override
    public void stop()
    {
      try {
        super.stop();
      } finally {
        try {
          if (client != null) {
            eventloop.disconnect(client);
            client = null;
          }
  
          eventloop.stop(server);
          eventloop.stop();
  
          if (codec instanceof Component) {
            @SuppressWarnings("unchecked")
            Component<com.datatorrent.api.Context> component = (Component<com.datatorrent.api.Context>)codec;
            component.teardown();
          }
          if (discovery instanceof Component) {
            @SuppressWarnings("unchecked")
            Component<com.datatorrent.api.Context> component = (Component<com.datatorrent.api.Context>)discovery;
            component.teardown();
          }
          if (storage instanceof Component) {
            @SuppressWarnings("unchecked")
            Component<com.datatorrent.api.Context> component = (Component<com.datatorrent.api.Context>)storage;
            component.teardown();
          }
        } catch (Throwable cause) {
          throw new ServiceConfigurationError("Failed Stop", cause);
        }
      }
    }
  
    /* End implementing Flume Sink interface */
  
    /* Begin Configurable Interface */
    @Override
    public void configure(Context context)
    {
      hostname = context.getString(HOSTNAME_STRING, HOSTNAME_DEFAULT);
      port = context.getInteger("port", 0);
      id = context.getString("id");
      if (id == null) {
        id = getName();
      }
      acceptedTolerance = context.getLong("acceptedTolerance", ACCEPTED_TOLERANCE);
      sleepMillis = context.getLong("sleepMillis", 5L);
      throughputAdjustmentFactor = context.getInteger("throughputAdjustmentPercent", 5) / 100.0;
      maximumEventsPerTransaction = context.getInteger("maximumEventsPerTransaction", 10000);
      minimumEventsPerTransaction = context.getInteger("minimumEventsPerTransaction", 100);
      commitEventTimeoutMillis = context.getLong("commitEventTimeoutMillis", Long.MAX_VALUE);
  
      @SuppressWarnings("unchecked")
      Discovery<byte[]> ldiscovery = configure("discovery", Discovery.class, context);
      if (ldiscovery == null) {
        logger.warn("Discovery agent not configured for the sink!");
        discovery = new Discovery<byte[]>()
        {
          @Override
          public void unadvertise(Service<byte[]> service)
          {
            logger.debug("Sink {} stopped listening on {}:{}", service.getId(), service.getHost(), service.getPort());
          }
  
          @Override
          public void advertise(Service<byte[]> service)
          {
            logger.debug("Sink {} started listening on {}:{}", service.getId(), service.getHost(), service.getPort());
          }
  
          @Override
          @SuppressWarnings("unchecked")
          public Collection<Service<byte[]>> discover()
          {
            return Collections.EMPTY_SET;
          }
  
        };
      } else {
        discovery = ldiscovery;
      }
  
      storage = configure("storage", Storage.class, context);
      if (storage == null) {
        logger.warn("storage key missing... FlumeSink may lose data!");
        storage = new Storage()
        {
          @Override
          public byte[] store(Slice slice)
          {
            return null;
          }
  
          @Override
          public byte[] retrieve(byte[] identifier)
          {
            return null;
          }
  
          @Override
          public byte[] retrieveNext()
          {
            return null;
          }
  
          @Override
          public void clean(byte[] identifier)
          {
          }
  
          @Override
          public void flush()
          {
          }
  
        };
      }
  
      @SuppressWarnings("unchecked")
      StreamCodec<Event> lCodec = configure("codec", StreamCodec.class, context);
      if (lCodec == null) {
        codec = new EventCodec();
      } else {
        codec = lCodec;
      }
  
    }
  
    /* End Configurable Interface */
  
    @SuppressWarnings({"UseSpecificCatch", "BroadCatchBlock", "TooBroadCatch"})
    private static <T> T configure(String key, Class<T> clazz, Context context)
    {
      String classname = context.getString(key);
      if (classname == null) {
        return null;
      }
  
      try {
        Class<?> loadClass = Thread.currentThread().getContextClassLoader().loadClass(classname);
        if (clazz.isAssignableFrom(loadClass)) {
          @SuppressWarnings("unchecked")
          T object = (T)loadClass.newInstance();
          if (object instanceof Configurable) {
            Context context1 = new Context(context.getSubProperties(key + '.'));
            String id = context1.getString(Storage.ID);
            if (id == null) {
              id = context.getString(Storage.ID);
              logger.debug("{} inherited id={} from sink", key, id);
              context1.put(Storage.ID, id);
            }
            ((Configurable)object).configure(context1);
          }
  
          return object;
        } else {
          logger.error("key class {} does not implement {} interface", classname, Storage.class.getCanonicalName());
          throw new Error("Invalid storage " + classname);
        }
      } catch (Error error) {
        throw error;
      } catch (RuntimeException re) {
        throw re;
      } catch (Throwable t) {
        throw new RuntimeException(t);
      }
    }
  
    /**
     * @return the hostname
     */
    String getHostname()
    {
      return hostname;
    }
  
    /**
     * @param hostname the hostname to set
     */
    void setHostname(String hostname)
    {
      this.hostname = hostname;
    }
  
    /**
     * @return the port
     */
    int getPort()
    {
      return port;
    }
  
    public long getAcceptedTolerance()
    {
      return acceptedTolerance;
    }
  
    public void setAcceptedTolerance(long acceptedTolerance)
    {
      this.acceptedTolerance = acceptedTolerance;
    }
  
    /**
     * @param port the port to set
     */
    void setPort(int port)
    {
      this.port = port;
    }
  
    /**
     * @return the discovery
     */
    Discovery<byte[]> getDiscovery()
    {
      return discovery;
    }
  
    /**
     * @param discovery the discovery to set
     */
    void setDiscovery(Discovery<byte[]> discovery)
    {
      this.discovery = discovery;
    }
  
    /**
     * Attempt the sequence of writing after sleeping twice and upon failure assume
     * that the client connection has problems and hence close it.
     *
     * @param address
     * @param e
     * @throws IOException
     */
    private void retryWrite(byte[] address, Slice event) throws IOException
    {
      if (event == null) {  /* this happens for playback where address and event are sent as single object */
        while (client.isConnected()) {
          sleep();
          if (client.write(address)) {
            return;
          }
        }
      } else {  /* this happens when the events are taken from the flume channel and writing first time failed */
        while (client.isConnected()) {
          sleep();
          if (client.write(address, event)) {
            return;
          }
        }
      }
  
      throw new IOException("Client disconnected!");
    }
  
    private static final Logger logger = LoggerFactory.getLogger(FlumeSink.class);
  }
  