public class CompoundKeyInfo implements Serializable {
    List<KeyInfo> keyInfoList = new ArrayList<KeyInfo>();
    int partitionId = -1;
    String compoundValue;
    String compoundKey;

    public CompoundKeyInfo() {
    }

    public void addKeyInfo(KeyInfo keyInfo) {
        keyInfoList.add(keyInfo);
    }

    public void setPartitionId(int partitionId) {
        this.partitionId = partitionId;
    }

    public void setCompoundKey(String compoundKey) {
        this.compoundKey = compoundKey;
    }

    public void setCompoundValue(String compoundValue) {
        this.compoundValue = compoundValue;
    }

    public List<KeyInfo> getKeyInfoList() {
        return keyInfoList;
    }

    public int getPartitionId() {
        return partitionId;
    }

    public String getCompoundKey() {
        if (compoundKey == null) {
            StringBuffer compoundKeyBuffer = new StringBuffer();
            for (KeyInfo keyInfo : this.getKeyInfoList()) {
                if (compoundKeyBuffer.length() > 0) {
                    compoundKeyBuffer.append(",");
                }
                compoundKeyBuffer.append(keyInfo.toString());
            }
            compoundKey = compoundKeyBuffer.toString();
        }
        return compoundKey;
    }

    public String getCompoundValue() {
        return this.compoundValue;
    }

    public String toString() {
        return "{" + getCompoundKey() + " = " + getCompoundValue() + "}:"
                + getPartitionId();
    }
} public class KeyInfo implements Serializable {
    List<KeyPathElement> keyPath = new ArrayList<KeyPathElement>();
    String value;

    public void setValue(String value) {
        this.value = value;
    }

    public String getValue() {
        return this.value;
    }

    public void addElementToPath(String keyName) {
        keyPath.add(new KeyPathElementName(keyName));
    }

    public void addElementToPath(int index) {
        keyPath.add(new KeyPathElementIndex(index));
    }

    private void addElementToPath(KeyPathElement keyPathElement) {
        keyPath.add(keyPathElement);
    }

    public List<KeyPathElement> getKeyPath() {
        return keyPath;
    }

    public static class KeyPathElement implements Serializable {
        public enum PathElementType {
            KEY_NAME, INDEX
        }

        PathElementType pathElementType;

        public PathElementType getPathElementType() {
            return pathElementType;
        }

    }

    public static class KeyPathElementName extends KeyPathElement {
        String keyName;

        public KeyPathElementName() {

        }

        public KeyPathElementName(String keyName) {
            pathElementType = PathElementType.KEY_NAME;
            this.keyName = keyName;
        }

        public String getKeyName() {
            return keyName;
        }
    }

    public static class KeyPathElementIndex extends KeyPathElement {
        int index;

        public KeyPathElementIndex() {

        }

        public KeyPathElementIndex(int index) {
            pathElementType = PathElementType.INDEX;
            this.index = index;
        }

        public int getIndex() {
            return index;
        }
    }

    public KeyInfo copy() {
        KeyInfo newKeyInfo = new KeyInfo();
        for (KeyPathElement element : keyPath) {
            newKeyInfo.addElementToPath(element);
        }
        return newKeyInfo;
    }

    public String toString() {
        return toString(false);
    }

    public String toString(boolean showFull) {
        StringBuffer sb = new StringBuffer();
        for (KeyPathElement element : keyPath) {
            if (element.getPathElementType() == KeyPathElement.PathElementType.KEY_NAME) {
                if (sb.length() > 0) {
                    sb.append("/");
                }
                sb.append(((KeyPathElementName) element).getKeyName());
            } else if (showFull) {
                sb.append("[")
                  .append(((KeyPathElementIndex) element).getIndex())
                  .append("]");
            }
        }
        return sb.toString();
    }
} public static class KeyPathElement implements Serializable {
    public enum PathElementType {
        KEY_NAME, INDEX
    }

    PathElementType pathElementType;

    public PathElementType getPathElementType() {
        return pathElementType;
    }

} public static class KeyPathElementIndex extends KeyPathElement {
    int index;

    public KeyPathElementIndex() {

    }

    public KeyPathElementIndex(int index) {
        pathElementType = PathElementType.INDEX;
        this.index = index;
    }

    public int getIndex() {
        return index;
    }
} public static class KeyPathElementName extends KeyPathElement {
    String keyName;

    public KeyPathElementName() {

    }

    public KeyPathElementName(String keyName) {
        pathElementType = PathElementType.KEY_NAME;
        this.keyName = keyName;
    }

    public String getKeyName() {
        return keyName;
    }
} public interface Persister {
    /**
     * Returns the number of <code>setAsynch</code>-initiated operations that
     * are pending or in-progress.
     * 
     * @return the number of pending set operations
     **/
    int getQueueSize();

    /**
     * Returns the number of <code>set</code> or <code>setAsynch</code> calls
     * made to this object.
     * 
     * @return the number of <code>set</code> or <code>setAsynch</code> calls
     *         made to this object.
     **/
    int getPersistCount();

    /**
     * Returns the number of key/value pairs in this Persister object.
     * <p>
     * If the underlying technology (e.g., Memcached) does not support this
     * operation, then the implementation may return 0 or a
     * <code>java.lang.UnsupportedOperationException</code>.
     * <p>
     * This number may include expired entries, depending on the implementation.
     * Some implementations clean up expired entries only when the value is
     * requested or at certain time intervals (or not at all).
     * 
     * @return the number of entries in this Persister
     **/
    int getCacheEntryCount();

    /**
     * Set a value for a key without waiting for the operation to complete.
     * <p>
     * This method is useful in the case where the underlying implementation
     * uses a network but the caller should not be delayed by the network
     * operation.
     * <p>
     * Because the set operation is not neccessarily complete when the method
     * returns, a subsequent <code>get</code> call may return stale data.
     * <p>
     * <code>period</code> specifies the time-to-live for this key/value pair.
     * For calls to <code>get</code> or <code>getObject</code> after the
     * specified period, the Persister object will return null.
     * 
     * @param key
     *            the key
     * @param value
     *            the value
     * @param period
     *            the maximum interval of time during which this key/value pair
     *            is valid. Also know as time-to-live. -1 indicates an infinite
     *            time-to-live.
     **/
    void setAsynch(String key, Object value, int period);

    /**
     * Set a value for a key.
     * <p>
     * <code>period</code> specifies the time-to-live for this key/value pair.
     * For calls to <code>get</code> or <code>getObject</code> after the
     * specified period, the Persister object will return null.
     * 
     * @param key
     *            the key
     * @param value
     *            the value
     * @param period
     *            the maximum interval of time during which this key/value pair
     *            is valid. Also know as time-to-live. -1 indicates an infinite
     *            time-to-live.
     **/
    void set(String key, Object value, int period) throws InterruptedException;

    /**
     * Get the value associated with a specified key.
     * 
     * If the period (aka time-to-live) for this key/value pair is expired, the
     * <code>get</code> will return null.
     * 
     * @param key
     *            the key
     **/
    Object get(String key) throws InterruptedException;

    /**
     * Get the values associated with a list of keys.
     * <p>
     * <code>getBulk</code> returns a <code>Map</code> containing an entry for
     * each key/value pair. The map contains an entry for a specified key only
     * if that key exists in the Persister and the associated key/value pair is
     * not yet expired.
     * 
     * @param keys
     *            a list of keys
     **/
    Map<String, Object> getBulk(String[] keys) throws InterruptedException;

    /**
     * This is a method to help support some implementations whose underlying
     * technology encodes the value associated with a key. In some cases, the
     * value may have been stored by a client that uses a form of encoding not
     * supported by the Persister's implementation. In that case, one might want
     * the raw, yet-to-be-decoded value associated with the key.
     * <p>
     * <code>getObject</code> retrieves the raw, unencoded value associated with
     * a key. In all other respects, it's the same as {@link Persister#get}.
     * <p>
     * It's likely you will never need to call this method.
     * 
     * If the period (aka time-to-live) for this key/value pair is expired, the
     * <code>get</code> will return null.
     **/
    Object getObject(String key) throws InterruptedException;

    /**
     * As with {@link Persister#getObject}, this is a method to help support
     * some implementations whose underlying technology encodes result.
     * <p>
     * It's likely you will never need to call this method.
     * <p>
     * In all other respects, it is the same as {@link Persister#getBulk}.
     **/
    Map<String, Object> getBulkObjects(String[] keys)
            throws InterruptedException;

    /**
     * Removes the entry for the specified key from the Persister object.
     * 
     * @param key
     *            the key of the entry to be removed
     **/
    void remove(String key) throws InterruptedException;

    /**
     * Forces the initiation of the process that cleans up expired entries.
     * <p>
     * Normally, you would not call this method. The implementation takes care
     * of the timing of the cleanup process.
     * <p>
     * In some implementations, this method may be a no-op.
     * 
     **/
    int cleanOutGarbage() throws InterruptedException;

    /**
     * Returns a <code>Set</code> of the keys contained in this Persister
     * object.
     * <p>
     * The <code>Set</code> may contain keys for expired entries, depending on
     * how often the underlying implementation cleans up expired entries.
     * <p>
     * The underlying technology may not support this operation. As a result,
     * some implementations return an empty set.
     * 
     * @return a <code>Set</code> of the keys contained in this Persister
     *         object.
     **/
    Set<String> keySet();
} public class Schema {
    private Map<String, Property> properties = new HashMap<String, Property>();
    @SuppressWarnings("unchecked")
    private Class type;

    public Map<String, Property> getProperties() {
        return Collections.unmodifiableMap(properties);
    }

    @SuppressWarnings("unchecked")
    public Class getType() {
        return type;
    }

    @SuppressWarnings("unchecked")
    public Schema(Class clazz) {
        type = clazz;
        for (Method method : clazz.getMethods()) {
            if (!method.getName().startsWith("set")
		|| !method.getReturnType().equals(Void.TYPE)
		|| method.getParameterTypes().length != 1) {
                continue;
            }

            String getterMethodName = method.getName().replaceFirst("set",
                                                                    "get");

            String propertyName = "";
            if (method.getName().length() > 4) {
                propertyName = method.getName().substring(4);
            }
            propertyName = method.getName().substring(3, 4).toLowerCase()
                    + propertyName;

            Type parameterType = method.getGenericParameterTypes()[0];

            Method getterMethod = null;
            try {
                getterMethod = clazz.getMethod(getterMethodName, new Class[] {});
                /*
                 * if (!getterMethod.getReturnType().equals(parameterType)) {
                 * getterMethod = null; }
                 */
            } catch (NoSuchMethodException nsme) {
                // this is an acceptable possibility, ignore
                // nsme.printStackTrace();
            }

            Property property = Property.getProperty(propertyName,
                                                     getterMethod,
                                                     method,
                                                     parameterType);
            if (property != null) {
                properties.put(propertyName, property);
            }
        }
    }

    public String toString() {
        return this.toString("");
    }

    public String toString(String indent) {
        StringBuffer sb = new StringBuffer();
        sb.append(indent).append("{\n");
        for (Property property : properties.values()) {
            sb.append(property.toString(indent + "   "));
        }

        sb.append(indent).append("}\n");
        return sb.toString();
    }

    static public class Property {
        @SuppressWarnings("unchecked")
        private static List<Class> nonBeanClasses = new ArrayList<Class>();
        static {
            nonBeanClasses.add(String.class);
            nonBeanClasses.add(Number.class);
        }
        @SuppressWarnings("unchecked")
        private Class type;
        private Property componentProperty;
        private Schema schema;
        private String name;
        private Method getterMethod;
        private Method setterMethod;
        private boolean isList;
        private boolean isNumber;

        @SuppressWarnings("unchecked")
        public Class getType() {
            return type;
        }

        public Schema getSchema() {
            return schema;
        }

        public String getName() {
            return name;
        }

        public Method getGetterMethod() {
            return getterMethod;
        }

        public Method getSetterMethod() {
            return setterMethod;
        }

        public boolean isList() {
            return isList;
        }

        public boolean isNumber() {
            return isNumber;
        }

        public Property getComponentProperty() {
            return componentProperty;
        }

        public static Property getProperty(String propertyName, Method getterMethod, Method setterMethod, Type parameterType) {
            if (parameterType instanceof Class) {
                Class ptClass = (Class) parameterType;

                if (ptClass.isArray()) {
                    Class componentType = ptClass.getComponentType();
                    return new Property(propertyName,
                                        getterMethod,
                                        setterMethod,
                                        ptClass,
                                        getProperty("component",
                                                    null,
                                                    null,
                                                    componentType));
                } else if (ptClass.isPrimitive()) {
                    return new Property(propertyName,
                                        getterMethod,
                                        setterMethod,
                                        ptClass);
                }

                List<Class> hierarchy = getHieararchy(ptClass);

                for (Class nonBeanClass : nonBeanClasses) {
                    if (hierarchy.contains(nonBeanClass)) {
                        return new Property(propertyName,
                                            getterMethod,
                                            setterMethod,
                                            ptClass);
                    }
                }

                // if here, then must have no-arg constructor and some bean-like
                // properties
                boolean noArgConstructorFound = false;
                for (Constructor constructor : ptClass.getConstructors()) {
                    if (constructor.getGenericParameterTypes().length == 0) {
                        noArgConstructorFound = true;
                        break;
                    }
                }

                if (!noArgConstructorFound) {
                    return null;
                }

                Schema schema = new Schema(ptClass);
                if (schema.getProperties().size() == 0) {
                    return null;
                }

                return new Property(propertyName,
                                    getterMethod,
                                    setterMethod,
                                    ptClass,
                                    schema);
            } else if (parameterType instanceof ParameterizedType) {
                ParameterizedType pt = (ParameterizedType) parameterType;
                Class rawType = (Class) pt.getRawType();

                Class[] interfaces = rawType.getInterfaces();
                if (!rawType.equals(List.class)
                        && !Arrays.asList(interfaces).contains(List.class)) {
                    return null;
                }

                Type[] parameterInstantiations = pt.getActualTypeArguments();
                if (parameterInstantiations.length != 1) {
                    return null;
                }

                Type pIType = parameterInstantiations[0];

                return new Property(propertyName,
                                    getterMethod,
                                    setterMethod,
                                    rawType,
                                    getProperty("component", null, null, pIType));
            }

            return null;
        }

        private static List<Class> getHieararchy(Class clazz) {
            List<Class> hierarchy = new ArrayList<Class>();

            for (Class level = clazz; level != null; level = level.getSuperclass()) {
                hierarchy.add(level);
            }
            return hierarchy;
        }

        private Property(String name, Method getterMethod, Method setterMethod,
                Class type) {
            this.name = name;
            this.getterMethod = getterMethod;
            this.setterMethod = setterMethod;
            this.type = type;

            Class[] interfaces = type.getInterfaces();
            if (type.equals(List.class)
                    || Arrays.asList(interfaces).contains(List.class)) {
                isList = true;
            } else if (getHieararchy(type).contains(Number.class)) {
                isNumber = true;
            }
        }

        private Property(String name, Method getterMethod, Method setterMethod,
                Class type, Schema schema) {
            this(name, getterMethod, setterMethod, type);
            this.schema = schema;
        }

        private Property(String name, Method getterMethod, Method setterMethod,
                Class type, Property componentProperty) {
            this(name, getterMethod, setterMethod, type);
            this.componentProperty = componentProperty;
        }

        public String toString() {
            return toString("");
        }

        public String toString(String indent) {
            StringBuffer sb = new StringBuffer();
            sb.append(indent)
              .append(type.getName())
              .append(" ")
              .append(name)
              .append("\n");
            if (schema != null) {
                sb.append(schema.toString(indent));
            }
            if (componentProperty != null) {
                sb.append(componentProperty.toString(indent));
            }

            return sb.toString();
        }
    }
} static public class Property {
    @SuppressWarnings("unchecked")
    private static List<Class> nonBeanClasses = new ArrayList<Class>();
    static {
        nonBeanClasses.add(String.class);
        nonBeanClasses.add(Number.class);
    }
    @SuppressWarnings("unchecked")
    private Class type;
    private Property componentProperty;
    private Schema schema;
    private String name;
    private Method getterMethod;
    private Method setterMethod;
    private boolean isList;
    private boolean isNumber;

    @SuppressWarnings("unchecked")
    public Class getType() {
        return type;
    }

    public Schema getSchema() {
        return schema;
    }

    public String getName() {
        return name;
    }

    public Method getGetterMethod() {
        return getterMethod;
    }

    public Method getSetterMethod() {
        return setterMethod;
    }

    public boolean isList() {
        return isList;
    }

    public boolean isNumber() {
        return isNumber;
    }

    public Property getComponentProperty() {
        return componentProperty;
    }

    public static Property getProperty(String propertyName, Method getterMethod, Method setterMethod, Type parameterType) {
        if (parameterType instanceof Class) {
            Class ptClass = (Class) parameterType;

            if (ptClass.isArray()) {
                Class componentType = ptClass.getComponentType();
                return new Property(propertyName,
                                    getterMethod,
                                    setterMethod,
                                    ptClass,
                                    getProperty("component",
                                                null,
                                                null,
                                                componentType));
            } else if (ptClass.isPrimitive()) {
                return new Property(propertyName,
                                    getterMethod,
                                    setterMethod,
                                    ptClass);
            }

            List<Class> hierarchy = getHieararchy(ptClass);

            for (Class nonBeanClass : nonBeanClasses) {
                if (hierarchy.contains(nonBeanClass)) {
                    return new Property(propertyName,
                                        getterMethod,
                                        setterMethod,
                                        ptClass);
                }
            }

            // if here, then must have no-arg constructor and some bean-like
            // properties
            boolean noArgConstructorFound = false;
            for (Constructor constructor : ptClass.getConstructors()) {
                if (constructor.getGenericParameterTypes().length == 0) {
                    noArgConstructorFound = true;
                    break;
                }
            }

            if (!noArgConstructorFound) {
                return null;
            }

            Schema schema = new Schema(ptClass);
            if (schema.getProperties().size() == 0) {
                return null;
            }

            return new Property(propertyName,
                                getterMethod,
                                setterMethod,
                                ptClass,
                                schema);
        } else if (parameterType instanceof ParameterizedType) {
            ParameterizedType pt = (ParameterizedType) parameterType;
            Class rawType = (Class) pt.getRawType();

            Class[] interfaces = rawType.getInterfaces();
            if (!rawType.equals(List.class)
                    && !Arrays.asList(interfaces).contains(List.class)) {
                return null;
            }

            Type[] parameterInstantiations = pt.getActualTypeArguments();
            if (parameterInstantiations.length != 1) {
                return null;
            }

            Type pIType = parameterInstantiations[0];

            return new Property(propertyName,
                                getterMethod,
                                setterMethod,
                                rawType,
                                getProperty("component", null, null, pIType));
        }

        return null;
    }

    private static List<Class> getHieararchy(Class clazz) {
        List<Class> hierarchy = new ArrayList<Class>();

        for (Class level = clazz; level != null; level = level.getSuperclass()) {
            hierarchy.add(level);
        }
        return hierarchy;
    }

    private Property(String name, Method getterMethod, Method setterMethod,
            Class type) {
        this.name = name;
        this.getterMethod = getterMethod;
        this.setterMethod = setterMethod;
        this.type = type;

        Class[] interfaces = type.getInterfaces();
        if (type.equals(List.class)
                || Arrays.asList(interfaces).contains(List.class)) {
            isList = true;
        } else if (getHieararchy(type).contains(Number.class)) {
            isNumber = true;
        }
    }

    private Property(String name, Method getterMethod, Method setterMethod,
            Class type, Schema schema) {
        this(name, getterMethod, setterMethod, type);
        this.schema = schema;
    }

    private Property(String name, Method getterMethod, Method setterMethod,
            Class type, Property componentProperty) {
        this(name, getterMethod, setterMethod, type);
        this.componentProperty = componentProperty;
    }

    public String toString() {
        return toString("");
    }

    public String toString(String indent) {
        StringBuffer sb = new StringBuffer();
        sb.append(indent)
          .append(type.getName())
          .append(" ")
          .append(name)
          .append("\n");
        if (schema != null) {
            sb.append(schema.toString(indent));
        }
        if (componentProperty != null) {
            sb.append(componentProperty.toString(indent));
        }

        return sb.toString();
    }
} public class SchemaContainer {
    private Map<String, Schema> schemaMap = new ConcurrentHashMap<String, Schema>();

    public Schema getSchema(Class<?> clazz) {
        String schemaKey = clazz.getClassLoader().toString() + ":"
                + clazz.getName();

        Schema schema = schemaMap.get(schemaKey);
        if (schema == null) {
            schema = new Schema(clazz);
            schemaMap.put(schemaKey, schema);
        }

        return schema;
    }

} public interface Clock {

    public long waitForTime(long targetTime);

    public long getCurrentTime();

}
import java.util.StringTokenizer;
import org.apache.log4j.Logger;
import io.s4.dispatcher.partitioner.CompoundKeyInfo;
import io.s4.dispatcher.partitioner.KeyInfo;
import io.s4.dispatcher.partitioner.KeyInfo.KeyPathElement;
import io.s4.dispatcher.partitioner.KeyInfo.KeyPathElementIndex;
import io.s4.dispatcher.partitioner.KeyInfo.KeyPathElementName;
import io.s4.persist.Persister;
import io.s4.schema.Schema;
import io.s4.schema.Schema.Property;
import io.s4.schema.SchemaContainer;
import io.s4.util.clock.Clock;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
/*
 * Copyright (c) 2010 Yahoo! Inc. All rights reserved.
 * 
 * Licensed 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. See accompanying LICENSE file. 
 */
package io.s4.processor;




/**
 * This is the base class for processor classes. While it is possible to create
 * a processor class by implementing the {@link ProcessingElement} interface, we
 * suggest you instead extend this class.
 * <p>
 * <code>AbstractProcessor</code> provides output frequency strategies that
 * allow you to configure the rate at which your processor produces output (see
 * {@link AbstractPE#setOutputFrequencyByEventCount} and
 * {@link AbstractPE#setOutputFrequencyByTimeBoundary}.
 */
public abstract class AbstractPE implements ProcessingElement {
    public static enum FrequencyType {
        TIMEBOUNDARY("timeboundary"), EVENTCOUNT("eventcount");

        private String name;

        FrequencyType(String name) {
            this.name = name;
        }

        public String getName() {
            return this.name;
        }
    }

    private Clock s4Clock;
    private int outputFrequency = 1;
    private FrequencyType outputFrequencyType = FrequencyType.EVENTCOUNT;
    private int outputFrequencyOffset = 0;
    private int eventCount = 0;
    private int ttl = -1;
    private Persister lookupTable;
    private List<EventAdvice> eventAdviceList = new ArrayList<EventAdvice>();
    private List<Object> keyValue;
    private List<Object> keyRecord;
    private String keyValueString;
    private String streamName;
    private boolean saveKeyRecord = false;
    private int outputsBeforePause = -1;
    private long pauseTimeInMillis;
    private boolean logPauses = false;
    private String initMethod = null;
    
    public void setSaveKeyRecord(boolean saveKeyRecord) {
        this.saveKeyRecord = saveKeyRecord;
    }

    public void setOutputsBeforePause(int outputsBeforePause) {
        this.outputsBeforePause = outputsBeforePause;
    }

    public void setPauseTimeInMillis(long pauseTimeInMillis) {
        this.pauseTimeInMillis = pauseTimeInMillis;
    }

    public void setLogPauses(boolean logPauses) {
        this.logPauses = logPauses;
    }

    public void setS4Clock(Clock s4Clock) {
        synchronized (this) {
            this.s4Clock = s4Clock;
            this.notify();
        }
    }

    /**
     * The name of a method to be used as an initializer.  The method will be
     * called after the object is cloned from the prototype PE.
     */
    public void setInitMethod(String initMethod)
    {
       this.initMethod = initMethod;
    }
    
    public String getInitMethod() {
       return this.initMethod;
    }
    
    public Clock getS4Clock() {
        return s4Clock;
    }

    private OverloadDispatcher overloadDispatcher;

    public AbstractPE() {
        OverloadDispatcherGenerator oldg = new OverloadDispatcherGenerator(this.getClass());
        Class<?> overloadDispatcherClass = oldg.generate();

        try {
            overloadDispatcher = (OverloadDispatcher) overloadDispatcherClass.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * This implements the <code>execute</code> method declared in the
     * {@link ProcessingElement} interface. You should not override this method.
     * Instead, you need to implement the <code>processEvent</code> method.
     **/
    public void execute(String streamName, CompoundKeyInfo compoundKeyInfo,
                        Object event) {
        // if this is the first time through, get the key for this PE
        if (keyValue == null || saveKeyRecord) {
            setKeyValue(event, compoundKeyInfo);

            if (compoundKeyInfo != null)
                keyValueString = compoundKeyInfo.getCompoundValue();
        }

        this.streamName = streamName;

        overloadDispatcher.dispatch(this, event);

        if (saveKeyRecord) {
            keyRecord.clear(); // the PE doesn't need it anymore
        }

        if (outputFrequencyType == FrequencyType.EVENTCOUNT
                && outputFrequency > 0) {
            eventCount++;
            if (eventCount % outputFrequency == 0) {
                try {
                    output();
                } catch (Exception e) {
                    Logger.getLogger("s4")
                          .error("Exception calling output() method in execute()",
                                 e);
                }
            }
        }
    }

    public long getCurrentTime() {
        return s4Clock.getCurrentTime();
    }

    /**
     * This method returns the key value associated with this PE.
     * <p>
     * The key value is a list because the key may be a compound (composite)
     * key, in which case the key will have one value for each simple key.
     * 
     * @return the key value as a List of Objects (each element contains the
     *         value of a simple key).
     **/
    public List<Object> getKeyValue() {
        return keyValue;
    }

    public List<Object> getKeyRecord() {
        return keyRecord;
    }

    public String getKeyValueString() {
        return keyValueString;
    }

    public String getStreamName() {
        return streamName;
    }

    SchemaContainer schemaContainer = new SchemaContainer();