class ArraySplitOperation implements MultiplexOperation {

  @Override
  public List<InternalEvent> perform(InternalEvent ievent) throws OperationException {
    {
      if (ievent.getEventObj() == null) {
        throw new OperationException("Deserialized object is null");
      }

      Object payload;
      try {
        payload = ievent.getEventObj().getField(this.path);
      } catch (FieldNotFoundException e) {
        throw new OperationException(e);
      }

      if (!(payload instanceof JsonArray)) {
        throw new OperationException("Payload data is not a JsonArray");
      }

      LinkedHashMap<String, String> partitions = ievent.getPartitions();

      JsonArray arr = (JsonArray) payload;

      ArrayList<InternalEvent> output = new ArrayList<InternalEvent>();
      for (JsonElement elm : arr) {
        try {
          JsonObject newObject = elm.getAsJsonObject();
          for (String field : this.fieldsToKeep) {
            JsonObject jsonObject = (JsonObject) ievent.getEventObj().getPayload();
            newObject.add(field, jsonObject.get(field));
          }

          InternalEvent newEvent = new InternalEvent(newObject.toString(), ievent.getCtx(), ievent.getArrivalTime());
          DeserializedEvent newDeserEvent = new GenericJsonEvent(newObject);
          newEvent.setEventObj(newDeserEvent);
          newEvent.setEventTime(ievent.getEventTime());

          /*
           * Deep clone the partitions
           */
          if (partitions != null) {
            LinkedHashMap<String, String> newPartitions =
                new LinkedHashMap<String, String>(partitions.size());

            partitions.entrySet().forEach(kv -> {
              newPartitions.put(new String(kv.getKey()), new String(kv.getValue()));
            });

            newEvent.setPartitions(newPartitions);
          }

          output.add(newEvent);
        } catch (Exception e) {
          throw new OperationException(e);
        }
      }

      return output;
    }
  }

  public  ArraySplitOperation(String path);
  public  ArraySplitOperation(String path,
                             List<String> fieldsToKeep);

}