Return-Path: 
 <user-return-2547-apmail-flink-user-archive=flink.apache.org@flink.apache.org>
X-Original-To: apmail-flink-user-archive@minotaur.apache.org
Delivered-To: apmail-flink-user-archive@minotaur.apache.org
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by minotaur.apache.org (Postfix) with SMTP id 2DCC518BCD
	for <apmail-flink-user-archive@minotaur.apache.org>;
 Tue,  1 Sep 2015 12:38:01 +0000 (UTC)
Received: (qmail 95151 invoked by uid 500); 1 Sep 2015 12:38:00 -0000
Delivered-To: apmail-flink-user-archive@flink.apache.org
Received: (qmail 95070 invoked by uid 500); 1 Sep 2015 12:38:00 -0000
Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:user-help@flink.apache.org>
List-Unsubscribe: <mailto:user-unsubscribe@flink.apache.org>
List-Post: <mailto:user@flink.apache.org>
List-Id: <user.flink.apache.org>
Reply-To: user@flink.apache.org
Delivered-To: mailing list user@flink.apache.org
Received: (qmail 95060 invoked by uid 99); 1 Sep 2015 12:38:00 -0000
Received: from Unknown (HELO spamd4-us-west.apache.org) (209.188.14.142)
    by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 01 Sep 2015 12:38:00 +0000
Received: from localhost (localhost [127.0.0.1])
	by spamd4-us-west.apache.org (ASF Mail Server at spamd4-us-west.apache.org)
 with ESMTP id 6E55BC0FA1
	for <user@flink.apache.org>; Tue,  1 Sep 2015 12:38:00 +0000 (UTC)
X-Virus-Scanned: Debian amavisd-new at spamd4-us-west.apache.org
X-Spam-Flag: NO
X-Spam-Score: 0.979
X-Spam-Level: 
X-Spam-Status: No, score=0.979 tagged_above=-999 required=6.31
	tests=[KAM_LAZY_DOMAIN_SECURITY=1, RCVD_IN_MSPIKE_H3=-0.01,
	RCVD_IN_MSPIKE_WL=-0.01, SPF_HELO_PASS=-0.001] autolearn=disabled
Received: from mx1-us-west.apache.org ([10.40.0.8])
	by localhost (spamd4-us-west.apache.org [10.40.0.11]) (amavisd-new,
 port 10024)
	with ESMTP id nROLjqIofC1o for <user@flink.apache.org>;
	Tue,  1 Sep 2015 12:37:57 +0000 (UTC)
Received: from smtprelay01.ispgateway.de (smtprelay01.ispgateway.de
 [80.67.29.23])
	by mx1-us-west.apache.org (ASF Mail Server at mx1-us-west.apache.org) with
 ESMTPS id CCE57205B8
	for <user@flink.apache.org>; Tue,  1 Sep 2015 12:37:56 +0000 (UTC)
Received: from [87.253.171.210] (helo=[10.90.158.38])
	by smtprelay01.ispgateway.de with esmtpsa (TLSv1.2:DHE-RSA-AES128-SHA:128)
	(Exim 4.84)
	(envelope-from <info@ricobergmann.de>)
	id 1ZWkov-0003FW-3u
	for user@flink.apache.org; Tue, 01 Sep 2015 14:37:49 +0200
To: user@flink.apache.org
From: "Dipl.-Inf. Rico Bergmann" <info@ricobergmann.de>
Subject: Duplicates in Flink
Message-ID: <55E59C1C.5090603@ricobergmann.de>
Date: Tue, 1 Sep 2015 14:37:48 +0200
User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:38.0) Gecko/20100101
 Thunderbird/38.2.0
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit
X-Df-Sender: aW5mb0ByaWNvYmVyZ21hbm4uZGU=

Hi!

I still have an issue... I was now using 0.9.1 and the new 
KafkaConnector. But I still get duplicates in my flink prog. Here's the 
relevant part:

         final FlinkKafkaConsumer082<String> kafkaSrc = new 
FlinkKafkaConsumer082<String>(
             kafkaTopicIn, new SimpleStringSchema(), properties);

         DataStream<String> start = env.addSource(kafkaSrc)
             .setParallelism(numReadPartitions); //numReadPartitions = 2

         DataStream<JSONObject> jsonized = start
             .flatMap(new ExtractAndFilterJSON());

         DataStream<Session> sessions = jsonized
             .partitionByHash(new KeySelector<JSONObject, String>() {
             /**
              * partition by session id
              */
             @Override
             public String getKey(JSONObject value)
                 throws Exception {
                 try {
                 return /*session id*/;
                 } catch (Exception e) {
                 LOG.error("no session could be retrieved", e);
                 }
                 return "";
             }
             }).flatMap(new StatefulSearchSessionizer());

In the StatefulSearchSessionizer I receive duplicates sporadically. I'm 
sure that the kafka topic I'm reading from does not contain any 
duplicates. So it must be in the flink program ...

Any ideas?

Cheers, Rico.