code: @Override
      public void update(EventBean[] newEvents, EventBean[] oldEvents) {
          if (newEvents == null) {
              return;
          }
          for (EventBean newEvent : newEvents) {
              List<Object> tuple = new ArrayList<>();
              String eventType;
              if (outputTypes.containsKey(newEvent.getEventType().getName())) {
                  eventType = newEvent.getEventType().getName();
                  for (String field : outputTypes.get(newEvent.getEventType().getName())) {
                      if (newEvent.get(field) != null) {
                          tuple.add(newEvent.get(field));
                      }
                  }
              } else {
                  eventType = "default";
                  for (String field : newEvent.getEventType().getPropertyNames()) {
                      tuple.add(newEvent.get(field));
                  }
              }
              collector.emit(eventType, tuple);
          }
      }
docs: {@inheritDoc}
  
  @param newEvents
  @param oldEvents