Java 类org.joda.time.Instant 实例源码

public static String buildBigQueryProcessedUrlsQuery(IndexerPipelineOptions options) {
    String timeWindow = null;

    if (options.getProcessedUrlHistorySec() != null) {
        if (options.getProcessedUrlHistorySec() != Integer.MAX_VALUE) {
            Instant fromTime = Instant.now();
            fromTime = fromTime.minus(options.getProcessedUrlHistorySec() * 1000L);
            Integer fromDateId = IdConverterUtils.getDateIdFromTimestamp(fromTime.getMillis());
            timeWindow = "PublicationDateId >= " + fromDateId;
        }
    }

    if (timeWindow != null)
        timeWindow = "WHERE " + timeWindow;

    String result = "SELECT Url, MAX(ProcessingTime) AS ProcessingTime\n" + "FROM " + options.getBigQueryDataset()
            + "." + WEBRESOURCE_TABLE + "\n" + timeWindow + "\n" + "GROUP BY Url";

    return result;
}

public static String buildBigQueryProcessedDocsQuery(IndexerPipelineOptions options) {
    String timeWindow = null;

    if (options.getProcessedUrlHistorySec() != null) {
        if (options.getProcessedUrlHistorySec() != Integer.MAX_VALUE) {
            Instant fromTime = Instant.now();
            fromTime = fromTime.minus(options.getProcessedUrlHistorySec() * 1000L);
            Integer fromDateId = IdConverterUtils.getDateIdFromTimestamp(fromTime.getMillis());
            timeWindow = "PublicationDateId >= " + fromDateId;
        }
    }

    if (timeWindow != null)
        timeWindow = "WHERE " + timeWindow;

    String result = "SELECT DocumentHash, MAX(ProcessingTime) AS ProcessingTime\n" + "FROM " + options.getBigQueryDataset()
            + "." + DOCUMENT_TABLE + "\n" + timeWindow + "\n" + "GROUP BY DocumentHash";

    return result;
}

public static String buildBigQueryProcessedSocialCountsQuery(IndexerPipelineOptions options) {
    String timeWindow = null;

    if (options.getWrSocialCountHistoryWindowSec() != null) {
        if (options.getWrSocialCountHistoryWindowSec() != Integer.MAX_VALUE) {
            Instant fromTime = Instant.now();
            fromTime = fromTime.minus(options.getWrSocialCountHistoryWindowSec() * 1000L);
            Integer fromDateId = IdConverterUtils.getDateIdFromTimestamp(fromTime.getMillis());
            timeWindow = "WrPublicationDateId >= " + fromDateId;
        }
    }

    if (timeWindow != null)
        timeWindow = "WHERE " + timeWindow;

    String result = "SELECT WebResourceHash, MAX(CountTime) AS LastCountTime\n" + "FROM "
            + options.getBigQueryDataset() + "." + WRSOCIALCOUNT_TABLE + "\n" + timeWindow + "\n"
            + "GROUP BY WebResourceHash";

    return result;

}

@ProcessElement
public void processElement(ProcessContext c) {
    WebresourceSocialCount sc = c.element();

    Instant countTime = new Instant(sc.countTime);

    TableRow row = new TableRow()
        .set("WebResourceHash", sc.webResourceHash)
        .set("WrPublicationDateId", sc.wrPublicationDateId)
        .set("CountTime", countTime.toString())
        .set("DocumentCollectionId", sc.documentCollectionId)
        .set("CollectionItemId", sc.collectionItemId)
        .set("FbCount", sc.fbCount)
        .set("TwCount", sc.twCount);

    c.output(row);

}

/**
 * input - a tupel that contains the data element (TableRow), the window, the timestamp, and the pane
 */

@Override
public TableDestination apply(ValueInSingleWindow<TableRow> input) {

    String partition;

    if (this.isTimeField) {
     String sTime = (String) input.getValue().get(this.fieldName);
     Instant time = Instant.parse(sTime);
     partition = time.toString(partitionFormatter);
    } else {
        partition = ((Integer) input.getValue().get(this.fieldName)).toString();
    }

    TableReference reference = new TableReference();
    reference.setProjectId(this.projectId);
    reference.setDatasetId(this.datasetId);
    reference.setTableId(this.partitionPrefix + partition);
    return new TableDestination(reference, null);
}

@Override
public PCollection<GameEvent> apply(PBegin begin) {
  if (options.getInput() != null && !options.getInput().isEmpty()) {
    return begin
        .getPipeline()
        .apply(TextIO.Read.from(options.getInput()))
        .apply(ParDo.named("ParseGameEvent").of(new ParseEventFn()))
        .apply(
            "AddEventTimestamps",
            WithTimestamps.of((GameEvent i) -> new Instant(i.getTimestamp())));
  } else {
    return begin
        .getPipeline()
        .apply(PubsubIO.Read.timestampLabel(TIMESTAMP_ATTRIBUTE).topic(options.getTopic()))
        .apply(ParDo.named("ParseGameEvent").of(new ParseEventFn()));
  }
}

@Test
public void testNotesWithTimesSkipDoneState() {
    shelfTestUtils.setupBook("notebook",
            "* Note 1\n"+
            "SCHEDULED: <2017-03-20>\n" +
            "* DONE Note 2\n"+
            "SCHEDULED: <2017-03-20>\n" +
            "* Note 3");

    ReminderService.LastRun lastRun = new ReminderService.LastRun();
    Instant now = Instant.parse("2017-03-15");
    AppPreferences.remindersForScheduledEnabled(context, true);

    List<NoteReminder> notes = ReminderService.getNoteReminders(
            context, now, lastRun, ReminderService.TIME_FROM_NOW);

    assertEquals(1, notes.size());
}

@Test
public void testNotesWithTimesWithRepeater() {
    shelfTestUtils.setupBook("notebook",
            "* Note 1\n"+
            "SCHEDULED: <2017-03-10 Fri +1w>\n" +
            "* Note 2\n"+
            "SCHEDULED: <2017-03-20 Mon 16:00>\n" +
            "* Note 3\n" +
            "* Note 4\n"+
            "SCHEDULED: <2017-03-16 Fri +1w>\n");

    ReminderService.LastRun lastRun = new ReminderService.LastRun();
    Instant now = Instant.parse("2017-03-15T13:00:00"); // Wed
    AppPreferences.remindersForScheduledEnabled(context, true);

    List<NoteReminder> notes = ReminderService.getNoteReminders(
            context, now, lastRun, ReminderService.TIME_FROM_NOW);

    assertEquals(2, notes.size());

    assertEquals("Note 4", notes.get(0).getPayload().title);
    assertEquals("2017-03-16T09:00:00", new LocalDateTime(notes.get(0).getRunTime()).toString("yyyy-MM-dd'T'HH:mm:ss"));

    assertEquals("Note 2", notes.get(1).getPayload().title);
    assertEquals("2017-03-20T16:00:00", new LocalDateTime(notes.get(1).getRunTime()).toString("yyyy-MM-dd'T'HH:mm:ss"));
}

@Test
public void testReminderForDeadlineTime() {
    shelfTestUtils.setupBook("notebook",
            "* Note 1\n"+
            "SCHEDULED: <2017-03-16 Fri +1w>\n" +
            "* Note 2\n"+
            "DEADLINE: <2017-03-20 Mon 16:00>\n");

    ReminderService.LastRun lastRun = new ReminderService.LastRun();
    Instant now = Instant.parse("2017-03-15T13:00:00"); // Wed
    AppPreferences.remindersForDeadlineEnabled(context, true);

    List<NoteReminder> notes = ReminderService.getNoteReminders(
            context, now, lastRun, ReminderService.TIME_FROM_NOW);

    assertEquals(1, notes.size());

    NoteReminder reminder = notes.get(0);
    assertEquals("Note 2", reminder.getPayload().title);
    assertEquals(DbTimeView.DEADLINE_TIME, reminder.getPayload().timeType);
    assertEquals("2017-03-20T16:00:00", new LocalDateTime(reminder.getRunTime()).toString("yyyy-MM-dd'T'HH:mm:ss"));
}

@Override
public void run(ApplicationArguments args) {
  Instant deletionCutoff = new Instant().minus(housekeeping.getExpiredPathDuration());
  LOG.info("Housekeeping at instant {} has started", deletionCutoff);
  CompletionCode completionCode = CompletionCode.SUCCESS;
  try {
    housekeepingService.cleanUp(deletionCutoff);
    LOG.info("Housekeeping at instant {} has finished", deletionCutoff);
  } catch (Exception e) {
    completionCode = CompletionCode.FAILURE;
    LOG.error("Housekeeping at instant {} has failed", deletionCutoff, e);
    throw e;
  } finally {
    Map<String, Long> metricsMap = ImmutableMap
        .<String, Long> builder()
        .put("housekeeping", completionCode.getCode())
        .build();
    metricSender.send(metricsMap);
  }
}

@Test
public void testApply() {
  DateTimeFormatter format = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
  Instant start = format.parseDateTime("2017-01-01 00:00:00").toInstant();
  Instant end = format.parseDateTime("2017-01-01 00:01:00").toInstant();
  IntervalWindow window = new IntervalWindow(start, end);

  String projectId = "testProject_id";
  String datasetId = "testDatasetId";
  String tablePrefix = "testTablePrefix";
  TableNameByWindowFn fn = new TableNameByWindowFn(projectId, datasetId, tablePrefix);
  String result = fn.apply(window);
  String expected = new Formatter()
      .format("%s:%s.%s_%s", projectId, datasetId, tablePrefix, "20170101")
      .toString();
  assertEquals(expected, result);
}

@Test
public void nonLatePanesMultiplePanes() {
  SerializableFunction<Iterable<ValueInSingleWindow<Integer>>, Iterable<Integer>> extractor =
      PaneExtractors.nonLatePanes();
  Iterable<ValueInSingleWindow<Integer>> onlyOnTime =
      ImmutableList.of(
          ValueInSingleWindow.of(
              8,
              new Instant(0L),
              GlobalWindow.INSTANCE,
              PaneInfo.createPane(false, false, Timing.LATE, 2L, 1L)),
          ValueInSingleWindow.of(7, new Instant(0L), GlobalWindow.INSTANCE, PaneInfo.NO_FIRING),
          ValueInSingleWindow.of(
              4,
              new Instant(0L),
              GlobalWindow.INSTANCE,
              PaneInfo.createPane(false, false, Timing.ON_TIME, 1L, 0L)),
          ValueInSingleWindow.of(
              1,
              new Instant(0L),
              GlobalWindow.INSTANCE,
              PaneInfo.createPane(true, false, Timing.EARLY)));

  assertThat(extractor.apply(onlyOnTime), containsInAnyOrder(4, 1, 7));
}

@Test
public void mutatedDuringProcessElementThrows() {
  WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes());
  CommittedBundle<byte[]> elements =
      bundleFactory.createBundle(pcollection).add(element).commit(Instant.now());

  ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer);
  enforcement.beforeElement(element);
  element.getValue()[0] = 'f';
  thrown.expect(IllegalMutationException.class);
  thrown.expectMessage(consumer.getFullName());
  thrown.expectMessage("illegaly mutated");
  thrown.expectMessage("Input values must not be mutated");
  enforcement.afterElement(element);
  enforcement.afterFinish(
      elements,
      StepTransformResult.<byte[]>withoutHold(consumer).build(),
      Collections.<CommittedBundle<?>>emptyList());
}

@Test
public void mutatedAfterProcessElementFails() {

  WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes());
  CommittedBundle<byte[]> elements =
      bundleFactory.createBundle(pcollection).add(element).commit(Instant.now());

  ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer);
  enforcement.beforeElement(element);
  enforcement.afterElement(element);

  element.getValue()[0] = 'f';
  thrown.expect(IllegalMutationException.class);
  thrown.expectMessage(consumer.getFullName());
  thrown.expectMessage("illegaly mutated");
  thrown.expectMessage("Input values must not be mutated");
  enforcement.afterFinish(
      elements,
      StepTransformResult.<byte[]>withoutHold(consumer).build(),
      Collections.<CommittedBundle<?>>emptyList());
}

@Test
public void testSynchronizedTimeMonotonic() {
  JavaSparkContext jsc = SparkContextFactory.getSparkContext(options);

  Instant instant = new Instant(0);
  GlobalWatermarkHolder.add(1,
      new SparkWatermarks(
          instant.plus(Duration.millis(5)),
          instant.plus(Duration.millis(10)),
          instant));
  GlobalWatermarkHolder.advance();

  thrown.expect(IllegalStateException.class);
  thrown.expectMessage("Synchronized processing time must advance.");
  // no actual advancement of watermarks - fine by Watermarks
  // but not by synchronized processing time.
  GlobalWatermarkHolder.add(1,
      new SparkWatermarks(
          instant.plus(Duration.millis(5)),
          instant.plus(Duration.millis(10)),
          instant));
  GlobalWatermarkHolder.advance();
}

@Test
public void mutationAfterAddKeyedBundleThrows() {
  UncommittedBundle<byte[]> keyed =
      factory.createKeyedBundle(StructuralKey.of("mykey", StringUtf8Coder.of()), transformed);

  byte[] array = new byte[] {4, 8, 12};
  WindowedValue<byte[]> windowedArray =
      WindowedValue.of(
          array,
          new Instant(891L),
          new IntervalWindow(new Instant(0), new Instant(1000)),
          PaneInfo.ON_TIME_AND_ONLY_FIRING);
  keyed.add(windowedArray);

  array[0] = Byte.MAX_VALUE;
  thrown.expect(IllegalMutationException.class);
  thrown.expectMessage("Values must not be mutated in any way after being output");
  keyed.commit(Instant.now());
}

@Test
public void allPanesMultiplePanes() {
  SerializableFunction<Iterable<ValueInSingleWindow<Integer>>, Iterable<Integer>> extractor =
      PaneExtractors.allPanes();
  Iterable<ValueInSingleWindow<Integer>> onlyOnTime =
      ImmutableList.of(
          ValueInSingleWindow.of(
              8,
              new Instant(0L),
              GlobalWindow.INSTANCE,
              PaneInfo.createPane(false, false, Timing.LATE, 2L, 1L)),
          ValueInSingleWindow.of(
              4,
              new Instant(0L),
              GlobalWindow.INSTANCE,
              PaneInfo.createPane(false, false, Timing.ON_TIME, 1L, 0L)),
          ValueInSingleWindow.of(
              1,
              new Instant(0L),
              GlobalWindow.INSTANCE,
              PaneInfo.createPane(true, false, Timing.EARLY)));

  assertThat(extractor.apply(onlyOnTime), containsInAnyOrder(4, 8, 1));
}

/**
 * Tests that when two elements are combined via a GroupByKey their output timestamp agrees
 * with the windowing function customized to actually be the same as the default, the earlier of
 * the two values.
 */
@Test
@Category(ValidatesRunner.class)
public void testTimestampCombinerEarliest() {

  p.apply(
      Create.timestamped(
          TimestampedValue.of(KV.of(0, "hello"), new Instant(0)),
          TimestampedValue.of(KV.of(0, "goodbye"), new Instant(10))))
      .apply(Window.<KV<Integer, String>>into(FixedWindows.of(Duration.standardMinutes(10)))
          .withTimestampCombiner(TimestampCombiner.EARLIEST))
      .apply(GroupByKey.<Integer, String>create())
      .apply(ParDo.of(new AssertTimestamp(new Instant(0))));

  p.run();
}

@Test
public void testAtWatermarkAndLate() throws Exception {
  tester = TriggerStateMachineTester.forTrigger(
      AfterWatermarkStateMachine.pastEndOfWindow()
          .withLateFirings(mockLate),
      FixedWindows.of(Duration.millis(100)));

  injectElements(1);
  IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(100));

  // No early firing, just double checking
  when(mockEarly.shouldFire(anyTriggerContext())).thenReturn(true);
  assertFalse(tester.shouldFire(window));
  tester.fireIfShouldFire(window);
  assertFalse(tester.isMarkedFinished(window));

  // Fire due to watermark
  when(mockEarly.shouldFire(anyTriggerContext())).thenReturn(false);
  tester.advanceInputWatermark(new Instant(100));
  assertTrue(tester.shouldFire(window));
  tester.fireIfShouldFire(window);
  assertFalse(tester.isMarkedFinished(window));

  testRunningAsTrigger(mockLate, window);
}

@Test
public void testIsWindowedValueReorderedWindows() {
  long timestamp = 100;
  long windowStart = 0;
  long windowEnd = 200;
  long windowStart2 = 50;
  long windowEnd2 = 150;

  assertThat(
      WindowedValue.of(
          "hello",
          new Instant(timestamp),
          ImmutableList.of(
              new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
              new IntervalWindow(new Instant(windowStart2), new Instant(windowEnd2))),
          PaneInfo.NO_FIRING),
      WindowMatchers.isWindowedValue(
          "hello",
          new Instant(timestamp),
          ImmutableList.of(
              new IntervalWindow(new Instant(windowStart), new Instant(windowEnd)),
              new IntervalWindow(new Instant(windowStart2), new Instant(windowEnd2))),
          PaneInfo.NO_FIRING));
}

@Test
public void shouldCountBytesWhenMultipleDataPointsReturned() throws Exception {
  Instant countSince = new Instant("2017-04-06T10:00:00.000Z");
  Instant countTo = new Instant("2017-04-06T11:00:00.000Z");
  Minutes periodTime = Minutes.minutesBetween(countSince, countTo);
  GetMetricStatisticsRequest metricStatisticsRequest =
      underTest.createMetricStatisticsRequest(STREAM, countSince, countTo, periodTime);
  GetMetricStatisticsResult result = new GetMetricStatisticsResult()
      .withDatapoints(
          new Datapoint().withSum(1.0),
          new Datapoint().withSum(3.0),
          new Datapoint().withSum(2.0)
      );

  given(cloudWatch.getMetricStatistics(metricStatisticsRequest)).willReturn(result);

  long backlogBytes = underTest.getBacklogBytes(STREAM, countSince, countTo);

  assertThat(backlogBytes).isEqualTo(6L);
}

@Test
public void testApplyIsScopedToExactClass() throws IOException {
  DataflowPipelineOptions options = buildPipelineOptions();
  Pipeline p = Pipeline.create(options);

  Create.TimestampedValues<String> transform =
      Create.timestamped(Arrays.asList(TimestampedValue.of("TestString", Instant.now())));
  p.apply(transform);

  CompositeTransformRecorder recorder = new CompositeTransformRecorder();
  p.traverseTopologically(recorder);

  // The recorder will also have seen a Create.Values composite as well, but we can't obtain that
  // transform.
  assertThat(
      "Expected to have seen CreateTimestamped composite transform.",
      recorder.getCompositeTransforms(),
      hasItem(transform));
  assertThat(
      "Expected to have two composites, CreateTimestamped and Create.Values",
      recorder.getCompositeTransforms(),
      hasItem(Matchers.<PTransform<?, ?>>isA((Class) Create.Values.class)));
}

@Test
public void testOnlyT1ShouldFireFixedWindows() throws Exception {
  tester =
      TriggerStateMachineTester.forTrigger(
          AfterFirstStateMachine.of(mockTrigger1, mockTrigger2),
          FixedWindows.of(Duration.millis(10)));
  tester.injectElements(1);
  IntervalWindow window = new IntervalWindow(new Instant(1), new Instant(11));

  when(mockTrigger1.shouldFire(anyTriggerContext())).thenReturn(true);
  when(mockTrigger2.shouldFire(anyTriggerContext())).thenReturn(false);

  assertTrue(tester.shouldFire(window)); // should fire

  tester.fireIfShouldFire(window);
  assertTrue(tester.isMarkedFinished(window));
}

@Override
protected void run() {
  TimestampedValue<AuctionBid> timestampedWinningBid = nextInput();
  if (timestampedWinningBid == null) {
    prune(NexmarkUtils.END_OF_TIME);
    for (TimestampedValue<CategoryPrice> result : lastSeenResults.values()) {
      addResult(result);
    }
    allDone();
    return;
  }
  lastTimestamp = timestampedWinningBid.getTimestamp();
  Instant newWindowStart = windowStart(Duration.standardSeconds(configuration.windowSizeSec),
      Duration.standardSeconds(configuration.windowPeriodSec), lastTimestamp);
  prune(newWindowStart);
  captureWinningBid(timestampedWinningBid.getValue().auction,
      timestampedWinningBid.getValue().bid, lastTimestamp);
}

@Test
public void fromCloudTimeShouldParseTimeStrings() {
  assertEquals(new Instant(0), fromCloudTime("1970-01-01T00:00:00Z"));
  assertEquals(new Instant(1), fromCloudTime("1970-01-01T00:00:00.001Z"));
  assertEquals(new Instant(1), fromCloudTime("1970-01-01T00:00:00.001000Z"));
  assertEquals(new Instant(1), fromCloudTime("1970-01-01T00:00:00.001001Z"));
  assertEquals(new Instant(1), fromCloudTime("1970-01-01T00:00:00.001000000Z"));
  assertEquals(new Instant(1), fromCloudTime("1970-01-01T00:00:00.001000001Z"));
  assertEquals(new Instant(0), fromCloudTime("1970-01-01T00:00:00.0Z"));
  assertEquals(new Instant(0), fromCloudTime("1970-01-01T00:00:00.00Z"));
  assertEquals(new Instant(420), fromCloudTime("1970-01-01T00:00:00.42Z"));
  assertEquals(new Instant(300), fromCloudTime("1970-01-01T00:00:00.3Z"));
  assertEquals(new Instant(20), fromCloudTime("1970-01-01T00:00:00.02Z"));
  assertNull(fromCloudTime(""));
  assertNull(fromCloudTime("1970-01-01T00:00:00"));
  assertNull(fromCloudTime("1970-01-01T00:00:00.1e3Z"));
}

@Override
public void fireTimer(Object key, Collection<TimerData> timerDataSet) {
  pushbackDoFnRunner.startBundle();
  @SuppressWarnings("unchecked")
  Coder<Object> keyCoder = (Coder) currentKeyStateInternals.getKeyCoder();
  ((StateInternalsProxy) currentKeyStateInternals).setKey(key);
  currentKeyTimerInternals.setContext(key, keyCoder, new Instant(this.currentInputWatermark),
      new Instant(this.currentOutputWatermark));
  for (TimerData timerData : timerDataSet) {
    StateNamespace namespace = timerData.getNamespace();
    checkArgument(namespace instanceof WindowNamespace);
    BoundedWindow window = ((WindowNamespace<?>) namespace).getWindow();
    pushbackDoFnRunner.onTimer(timerData.getTimerId(), window,
        timerData.getTimestamp(), timerData.getDomain());
  }
  pushbackDoFnRunner.finishBundle();
}

/**
 * Handle an account state warning produced by ldaptive account state machinery.
 * <p>
 * Override this method to provide custom warning message handling.
 *
 * @param warning the account state warning messages.
 * @param response Ldaptive authentication response.
 * @param configuration Password policy configuration.
 * @param messages Container for messages produced by account state warning handling.
 */
protected void handleWarning(
        final AccountState.Warning warning,
        final AuthenticationResponse response,
        final LdapPasswordPolicyConfiguration configuration,
        final List<MessageDescriptor> messages) {

    logger.debug("Handling warning {}", warning);
    if (warning == null) {
        logger.debug("Account state warning not defined");
        return;
    }

    final Calendar expDate = warning.getExpiration();
    final Days ttl = Days.daysBetween(Instant.now(), new Instant(expDate));
    logger.debug(
            "Password expires in {} days. Expiration warning threshold is {} days.",
            ttl.getDays(),
            configuration.getPasswordWarningNumberOfDays());
    if (configuration.isAlwaysDisplayPasswordExpirationWarning()
            || ttl.getDays() < configuration.getPasswordWarningNumberOfDays()) {
        messages.add(new PasswordExpiringWarningMessageDescriptor(
                "Password expires in {0} days. Please change your password at <href=\"{1}\">{1}</a>",
                ttl.getDays(),
                configuration.getPasswordPolicyUrl()));
    }
    if (warning.getLoginsRemaining() > 0) {
        messages.add(new DefaultMessageDescriptor(
                "password.expiration.loginsRemaining",
                "You have {0} logins remaining before you MUST change your password.",
                warning.getLoginsRemaining()));

    }
}

/**
 * Handle an account state warning produced by ldaptive account state machinery.
 * <p>
 * Override this method to provide custom warning message handling.
 *
 * @param warning the account state warning messages.
 * @param response Ldaptive authentication response.
 * @param configuration Password policy configuration.
 * @param messages Container for messages produced by account state warning handling.
 */
protected void handleWarning(
        final AccountState.Warning warning,
        final AuthenticationResponse response,
        final LdapPasswordPolicyConfiguration configuration,
        final List<MessageDescriptor> messages) {

    logger.debug("Handling warning {}", warning);
    if (warning == null) {
        logger.debug("Account state warning not defined");
        return;
    }

    final Calendar expDate = warning.getExpiration();
    final Days ttl = Days.daysBetween(Instant.now(), new Instant(expDate));
    logger.debug(
            "Password expires in {} days. Expiration warning threshold is {} days.",
            ttl.getDays(),
            configuration.getPasswordWarningNumberOfDays());
    if (configuration.isAlwaysDisplayPasswordExpirationWarning()
            || ttl.getDays() < configuration.getPasswordWarningNumberOfDays()) {
        messages.add(new PasswordExpiringWarningMessageDescriptor(
                "Password expires in {0} days. Please change your password at <href=\"{1}\">{1}</a>",
                ttl.getDays(),
                configuration.getPasswordPolicyUrl()));
    }
    if (warning.getLoginsRemaining() > 0) {
        messages.add(new DefaultMessageDescriptor(
                "password.expiration.loginsRemaining",
                "You have {0} logins remaining before you MUST change your password.",
                warning.getLoginsRemaining()));

    }
}

public EntityDeathEvent(Entity entity, Location location, Lifetime lifetime, Instant time, List<ItemStack> drops, int droppedExp) {
    super(entity);
    this.entity = entity;
    this.location = location;
    this.lifetime = lifetime;
    this.time = time;
    this.drops = drops;
    this.droppedExp = droppedExp;
}

public SimpleDamage(int damage, @Nonnull Location location, @Nonnull Instant time, @Nonnull DamageInfo info) {
    Preconditions.checkArgument(damage >= 0, "damage must be greater than or equal to zero");
    Preconditions.checkNotNull(location, "location");
    Preconditions.checkNotNull(time, "time");
    Preconditions.checkNotNull(info, "info");

    this.damage = damage;
    this.location = location;
    this.time = time;
    this.info = info;
}

public EntityDamageEvent(@Nonnull T entity, @Nonnull Lifetime lifetime, int damage, @Nonnull Location location, @Nonnull Instant time, @Nonnull DamageInfo info) {
    Preconditions.checkNotNull(entity, "entity");
    Preconditions.checkNotNull(lifetime, "lifetime");
    Preconditions.checkArgument(damage >= 0, "damage must be greater than or equal to zero");
    Preconditions.checkNotNull(location, "location");
    Preconditions.checkNotNull(time, "time");
    Preconditions.checkNotNull(info, "damage info");

    this.entity = entity;
    this.lifetime = lifetime;
    this.damage = damage;
    this.location = location.clone();
    this.time = time;
    this.info = info;
}

@ProcessElement
public void processElement(ProcessContext c) {
    KV<String,ContentIndexSummary> kv = c.element();
    String newDocumentHash = kv.getKey();
    ContentIndexSummary summary = kv.getValue();

    // Create the webresource entry
    Instant pubTime = new Instant(summary.wr.publicationTime);
    Instant proTime = new Instant(summary.wr.processingTime);

    // TODO: we are leaving summary.wr.collectionItemId, summary.wr.publicationDateId unchanged for now 
    // These values are different from the Document to which we repointed the WebResource
    // It could be a good or a bad thing, depending on the circumstances

    TableRow wrrow = new TableRow()
        .set("WebResourceHash", summary.wr.webResourceHash)
        .set("Url", summary.wr.url)
        .set("PublicationTime", pubTime.toString())
        .set("PublicationDateId", summary.wr.publicationDateId)
        .set("ProcessingTime", proTime.toString())
        .set("ProcessingDateId", summary.wr.processingDateId)
        .set("DocumentHash", newDocumentHash); // replace the original DocumentHash with the passed value

    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "DocumentCollectionId", summary.wr.documentCollectionId);
    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "CollectionItemId", summary.wr.collectionItemId);
    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "Title", summary.wr.title);
    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "Domain", summary.wr.domain);
    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "Author", summary.wr.author);
    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "ParentWebResourceHash", summary.wr.parentWebResourceHash);
    IndexerPipelineUtils.setTableRowFieldIfNotNull(wrrow, "MetaFields", summary.wr.metaFields);

    c.output(wrrow);
}

public static void main(String[] args) throws NoSuchAlgorithmException {

    SecureRandom sr = SecureRandom.getInstance("SHA1PRNG");
    for (int i=1; i<=1440; i++) {
        System.out.println("maps.put(\"20141109-"+i+"\", \""+sr.nextInt(10)+" "+sr.nextInt(10)+" "+sr.nextInt(10)+" "+sr.nextInt(10)+" "+sr.nextInt(10)+"\");");
    }

    System.out.println(LocalDate.now());
    Instant in = new Instant(1414508801016L);
    DateTimeFormatter formatter = DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss");
    formatter=formatter.withZone(DateTimeZone.forTimeZone(TimeZone.getTimeZone("GMT+8")));

    in = in.plus(100);

    System.out.println(in.get(DateTimeFieldType.millisOfSecond()));

    System.out.println(in.toDate());
    System.out.println(formatter.print(in));
    System.out.println(in.getMillis());
    Pattern pattern = Pattern.compile("\"phase\":\"20141018023\"(.*)\"data\":\\[\"(\\d)\",\"(\\d)\",\"(\\d)\",\"(\\d)\",\"(\\d)\"\\]\\}\\]\\}");
    Matcher matcher = pattern.matcher("{\"code\":0,\"message\":\"\",\"data\":[{\"phasetype\":200,\"phase\":\"20141018023\",\"create_at\":\"2014-01-21 14:41:05\",\"time_startsale\":\"2014-10-18 01:50:00\",\"time_endsale\":\"2014-10-18 01:55:00\",\"time_endticket\":\"2014-10-18 01:55:00\",\"time_draw\":\"2014-10-18 01:56:00\",\"status\":5,\"forsale\":0,\"is_current\":0,\"result\":{\"result\":[{\"key\":\"ball\",\"data\":[\"1\",\"5\",\"0\",\"5\",\"9\"]}]},\"result_detail\":{\"resultDetail\":[{\"key\":\"prize1\",\"bet\":\"0\",\"prize\":100000},{\"key\":\"prize2\",\"bet\":\"0\",\"prize\":20000},{\"key\":\"prize3\",\"bet\":\"0\",\"prize\":200},{\"key\":\"prize4\",\"bet\":\"0\",\"prize\":20},{\"key\":\"prize5\",\"bet\":\"0\",\"prize\":1000},{\"key\":\"prize6\",\"bet\":\"0\",\"prize\":320},{\"key\":\"prize7\",\"bet\":\"0\",\"prize\":160},{\"key\":\"prize8\",\"bet\":\"0\",\"prize\":100},{\"key\":\"prize9\",\"bet\":\"0\",\"prize\":50},{\"key\":\"prize10\",\"bet\":\"0\",\"prize\":10},{\"key\":\"prize11\",\"bet\":\"0\",\"prize\":4}]},\"pool_amount\":\"\",\"sale_amount\":\"\",\"ext\":\"\",\"fc3d_sjh\":null,\"terminal_status\":2,\"fordraw\":0,\"time_startsale_fixed\":\"2014-10-18 01:47:40\",\"time_endsale_fixed\":\"2014-10-18 01:52:40\",\"time_endsale_syndicate_fixed\":\"2014-10-18 01:55:00\",\"time_endsale_upload_fixed\":\"2014-10-18 01:55:00\",\"time_draw_fixed\":\"2014-10-18 01:56:00\",\"time_startsale_correction\":140,\"time_endsale_correction\":140,\"time_endsale_syndicate_correction\":0,\"time_endsale_upload_correction\":0,\"time_draw_correction\":0,\"time_exchange\":\"2014-12-16 01:56:00\"},{\"phasetype\":\"200\",\"phase\":\"20141018024\",\"create_at\":\"2014-01-21 14:41:05\",\"time_startsale\":\"2014-10-18 01:55:00\",\"time_endsale\":\"2014-10-18 10:00:00\",\"time_endticket\":\"2014-10-18 10:00:00\",\"time_draw\":\"2014-10-18 10:01:00\",\"status\":\"2\",\"forsale\":\"1\",\"is_current\":\"1\",\"result\":null,\"result_detail\":null,\"pool_amount\":\"\",\"sale_amount\":\"\",\"ext\":\"\",\"fc3d_sjh\":null,\"terminal_status\":\"1\",\"fordraw\":\"0\",\"time_startsale_fixed\":\"2014-10-18 01:52:40\",\"time_endsale_fixed\":\"2014-10-18 09:57:40\",\"time_endsale_syndicate_fixed\":\"2014-10-18 10:00:00\",\"time_endsale_upload_fixed\":\"2014-10-18 10:00:00\",\"time_draw_fixed\":\"2014-10-18 10:01:00\",\"time_startsale_correction\":140,\"time_endsale_correction\":140,\"time_endsale_syndicate_correction\":0,\"time_endsale_upload_correction\":0,\"time_draw_correction\":0,\"time_exchange\":\"2014-12-16 10:01:00\"}],\"redirect\":\"\",\"datetime\":\"2014-10-18 04:08:45\",\"timestamp\":1413576525}");

    //Pattern pattern = Pattern.compile("(.*)message(\\d\\d)(\\d)(\\d)(\\d)");
    //Matcher matcher = pattern.matcher("23fawef_message12345");
    //Pattern pattern = Pattern.compile("\"number\":\"(\\d) (\\d) (\\d) (\\d) (\\d)\",\"period\":\"20141017083");
    //Matcher matcher = pattern.matcher("{\"latestPeriods\":[{\"number\":\"6 0 2 2 1\",\"period\":\"20141017084\"},{\"number\":\"0 8 9 1 9\",\"period\":\"20141017083\"},{\"number\":\"4 0 4 4 6\",\"period\":\"20141017082\"},{\"number\":\"4 5 8 7 7\",\"period\":\"20141017081\"},{\"number\":\"7 2 8 5 3\",\"period\":\"20141017080\"},{\"number\":\"9 7 3 8 0\",\"period\":\"20141017079\"},{\"number\":\"3 7 6 0 1\",\"period\":\"20141017078\"},{\"number\":\"9 6 4 8 5\",\"period\":\"20141017077\"},{\"number\":\"6 4 1 8 1\",\"period\":\"20141017076\"},{\"number\":\"9 5 2 8 7\",\"period\":\"20141017075\"}],\"successful\":\"true\",\"statusDesc\":\"获取数据成功\"}");

    matcher.find();
    for (int i=1; i<=matcher.groupCount(); i++) {

            System.out.println(matcher.group(i));

    }
}

/**
 * Convert an extended bounds in parsed for into one in unparsed form.
 */
public static ExtendedBounds unparsed(ExtendedBounds template) {
    // It'd probably be better to randomize the formatter
    FormatDateTimeFormatter formatter = Joda.forPattern("dateOptionalTime");
    String minAsStr = template.getMin() == null ? null : formatter.printer().print(new Instant(template.getMin()));
    String maxAsStr = template.getMax() == null ? null : formatter.printer().print(new Instant(template.getMax()));
    return new ExtendedBounds(minAsStr, maxAsStr);
}

/**
 * Handle an account state warning produced by ldaptive account state machinery.
 * <p>
 * Override this method to provide custom warning message handling.
 *
 * @param error Account state warning.
 * @param response Ldaptive authentication response.
 * @param configuration Password policy configuration.
 * @param messages Container for messages produced by account state warning handling.
 */
protected void handleWarning(
        final AccountState.Warning warning,
        final AuthenticationResponse response,
        final LdapPasswordPolicyConfiguration configuration,
        final List<Message> messages) {

    if (warning == null) {
        logger.debug("Account state warning not defined");
        return;
    }

    final Calendar expDate = warning.getExpiration();
    final Days ttl = Days.daysBetween(Instant.now(), new Instant(expDate));
    logger.debug(
            "Password expires in {} days. Expiration warning threshold is {} days.",
            ttl.getDays(),
            configuration.getPasswordWarningNumberOfDays());
    if (configuration.isAlwaysDisplayPasswordExpirationWarning()
            || ttl.getDays() < configuration.getPasswordWarningNumberOfDays()) {
        messages.add(new PasswordExpiringWarningMessage(
                "Password expires in {0} days. Please change your password at <href=\"{1}\">{1}</a>",
                ttl.getDays(),
                configuration.getPasswordPolicyUrl()));
    }
    if (warning.getLoginsRemaining() > 0) {
        messages.add(new Message(
                "password.expiration.loginsRemaining",
                "You have {0} logins remaining before you MUST change your password.",
                warning.getLoginsRemaining()));

    }
}

@Override
public void processElement(ProcessContext c) {
  TableRow row =
      new TableRow()
          .set("team", c.element().getKey())
          .set("total_score", c.element().getValue())
          .set("window_start", ((IntervalWindow) c.window()).start().getMillis() / 1000)
          .set("processing_time", Instant.now().getMillis() / 1000)
          .set("timing", c.pane().getTiming().toString());
  c.output(row);
}

@Override
public void processElement(ProcessContext c) {
  TableRow row =
      new TableRow()
          .set("user", c.element().getKey())
          .set("total_score", c.element().getValue())
          .set("processing_time", Instant.now().getMillis() / 1000);
  c.output(row);
}

@Override
public void processElement(ProcessContext c) {
  TableRow row =
      new TableRow()
          .set("bad_user", c.element().getKey())
          .set("time", Instant.now().getMillis() / 1000);
  c.output(row);
}

@Override
public void processElement(ProcessContext c) {
  TableRow row =
      new TableRow()
          .set("team", c.element().getKey())
          .set("total_score", c.element().getValue())
          .set("window_start", ((IntervalWindow) c.window()).start().getMillis() / 1000)
          .set("processing_time", Instant.now().getMillis() / 1000)
          .set("timing", c.pane().getTiming().toString());
  c.output(row);
}

@Override
public void processElement(ProcessContext c) {
  TableRow row =
      new TableRow()
          .set("user", c.element().getKey())
          .set("total_score", c.element().getValue())
          .set("processing_time", Instant.now().getMillis() / 1000);
  c.output(row);
}