Return-Path: <commits-return-23078-archive-asf-public=cust-asf.ponee.io@beam.apache.org>
X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io
Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io
Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183])
	by cust-asf2.ponee.io (Postfix) with ESMTP id CE0A2200C22
	for <archive-asf-public-internal@cust-asf2.ponee.io>; Tue, 21 Feb 2017 14:51:54 +0100 (CET)
Received: by cust-asf.ponee.io (Postfix)
	id CCB59160B68; Tue, 21 Feb 2017 13:51:54 +0000 (UTC)
Delivered-To: archive-asf-public@cust-asf.ponee.io
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by cust-asf.ponee.io (Postfix) with SMTP id 13CF4160B3E
	for <archive-asf-public@cust-asf.ponee.io>; Tue, 21 Feb 2017 14:51:53 +0100 (CET)
Received: (qmail 19409 invoked by uid 500); 21 Feb 2017 13:51:53 -0000
Mailing-List: contact commits-help@beam.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:commits-help@beam.apache.org>
List-Unsubscribe: <mailto:commits-unsubscribe@beam.apache.org>
List-Post: <mailto:commits@beam.apache.org>
List-Id: <commits.beam.apache.org>
Reply-To: dev@beam.apache.org
Delivered-To: mailing list commits@beam.apache.org
Received: (qmail 19396 invoked by uid 99); 21 Feb 2017 13:51:53 -0000
Received: from pnap-us-west-generic-nat.apache.org (HELO spamd2-us-west.apache.org) (209.188.14.142)
    by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 21 Feb 2017 13:51:53 +0000
Received: from localhost (localhost [127.0.0.1])
	by spamd2-us-west.apache.org (ASF Mail Server at spamd2-us-west.apache.org) with ESMTP id B14921A7ACA
	for <commits@beam.apache.org>; Tue, 21 Feb 2017 13:51:52 +0000 (UTC)
X-Virus-Scanned: Debian amavisd-new at spamd2-us-west.apache.org
X-Spam-Flag: NO
X-Spam-Score: -1.999
X-Spam-Level: 
X-Spam-Status: No, score=-1.999 tagged_above=-999 required=6.31
	tests=[KAM_LAZY_DOMAIN_SECURITY=1, RP_MATCHES_RCVD=-2.999]
	autolearn=disabled
Received: from mx1-lw-eu.apache.org ([10.40.0.8])
	by localhost (spamd2-us-west.apache.org [10.40.0.9]) (amavisd-new, port 10024)
	with ESMTP id qnKCXiJSBDwF for <commits@beam.apache.org>;
	Tue, 21 Feb 2017 13:51:51 +0000 (UTC)
Received: from mailrelay1-us-west.apache.org (mailrelay1-us-west.apache.org [209.188.14.139])
	by mx1-lw-eu.apache.org (ASF Mail Server at mx1-lw-eu.apache.org) with ESMTP id 4986E5F4A8
	for <commits@beam.incubator.apache.org>; Tue, 21 Feb 2017 13:51:51 +0000 (UTC)
Received: from jira-lw-us.apache.org (unknown [207.244.88.139])
	by mailrelay1-us-west.apache.org (ASF Mail Server at mailrelay1-us-west.apache.org) with ESMTP id 4EA88E02F1
	for <commits@beam.incubator.apache.org>; Tue, 21 Feb 2017 13:51:44 +0000 (UTC)
Received: from jira-lw-us.apache.org (localhost [127.0.0.1])
	by jira-lw-us.apache.org (ASF Mail Server at jira-lw-us.apache.org) with ESMTP id 0BD892411F
	for <commits@beam.incubator.apache.org>; Tue, 21 Feb 2017 13:51:44 +0000 (UTC)
Date: Tue, 21 Feb 2017 13:51:44 +0000 (UTC)
From: "Jingsong Lee (JIRA)" <jira@apache.org>
To: commits@beam.incubator.apache.org
Message-ID: <JIRA.13044761.1487667730000.18592.1487685104046@Atlassian.JIRA>
In-Reply-To: <JIRA.13044761.1487667730000@Atlassian.JIRA>
References: <JIRA.13044761.1487667730000@Atlassian.JIRA> <JIRA.13044761.1487667730491@jira-lw-us.apache.org>
Subject: [jira] [Commented] (BEAM-1517) Garbage collect user state in Flink
 Runner
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
X-JIRA-FingerPrint: 30527f35849b9dde25b450d4833f0394
archived-at: Tue, 21 Feb 2017 13:51:55 -0000


    [ https://issues.apache.org/jira/browse/BEAM-1517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15876005#comment-15876005 ] 

Jingsong Lee commented on BEAM-1517:
------------------------------------

Is it appropriate for the user to do the work of GC?
Just like this:
{code}
  @ProcessElement
  public void process(
      ProcessContext c,
      BoundedWindow window,
      @StateId(stateId) ValueState<Integer> state,
      @TimerId("GcTimer") Timer timer) {
    Instant maxTimestamp = window.maxTimestamp();
    long allowedLateness = 10 * 1000;
    Instant gcTime = maxTimestamp.plus(allowedLateness);
    //Can Timer have a getCurrentTime interface?
    Instant currentTime = new Instant();
    if (gcTime.isBefore(currentTime)) {
      c.sideOutput(lateDataTag, c.element());
    } else {
      timer.set(gcTime);
      // user logical
      // ....
    }
  }
  @OnTimer("GcTimer")
  public void gc(
      OnTimerContext context,
      @StateId(stateId) ValueState<Integer> state) {
    state.clear();
  }
{code}


> Garbage collect user state in Flink Runner
> ------------------------------------------
>
>                 Key: BEAM-1517
>                 URL: https://issues.apache.org/jira/browse/BEAM-1517
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-flink
>            Reporter: Aljoscha Krettek
>            Assignee: Aljoscha Krettek
>
> User facing state/timers in Beam are bound to the key/window of the data. Right now, the Flink Runner does not clean up user state when the watermark passes the GC horizon for the state associated with a given window.
> Neither {{StateInternals}} nor the Flink state API support discarding state for a whole namespace (which is the window in this case) so we might have to manually set a GC timer for each window/key combination, as is done in the {{ReduceFnRunner}}. For this we have to know all states a user can possibly use, which we can get from the {{DoFn}} signature.



--
This message was sent by Atlassian JIRA
(v6.3.15#6346)