Return-Path: 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 2C03C200B62 for ; Fri, 29 Jul 2016 05:54:10 +0200 (CEST) Received: by cust-asf.ponee.io (Postfix) id 2AB31160A94; Fri, 29 Jul 2016 03:54:10 +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 4A56D160A85 for ; Fri, 29 Jul 2016 05:54:09 +0200 (CEST) Received: (qmail 68862 invoked by uid 500); 29 Jul 2016 03:54:08 -0000 Mailing-List: contact user-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@flink.apache.org Delivered-To: mailing list user@flink.apache.org Received: (qmail 68852 invoked by uid 99); 29 Jul 2016 03:54:08 -0000 Received: from pnap-us-west-generic-nat.apache.org (HELO spamd4-us-west.apache.org) (209.188.14.142) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 29 Jul 2016 03:54:08 +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 CC6E3C041B for ; Fri, 29 Jul 2016 03:54:07 +0000 (UTC) X-Virus-Scanned: Debian amavisd-new at spamd4-us-west.apache.org X-Spam-Flag: NO X-Spam-Score: 1.199 X-Spam-Level: * X-Spam-Status: No, score=1.199 tagged_above=-999 required=6.31 tests=[DKIM_SIGNED=0.1, DKIM_VALID=-0.1, DKIM_VALID_AU=-0.1, HTML_MESSAGE=2, RCVD_IN_DNSWL_LOW=-0.7, RCVD_IN_MSPIKE_H2=-0.001, SPF_PASS=-0.001, UNPARSEABLE_RELAY=0.001] autolearn=disabled Authentication-Results: spamd4-us-west.apache.org (amavisd-new); dkim=pass (2048-bit key) header.d=gmail.com Received: from mx2-lw-eu.apache.org ([10.40.0.8]) by localhost (spamd4-us-west.apache.org [10.40.0.11]) (amavisd-new, port 10024) with ESMTP id DFY5GBGBzsDz for ; Fri, 29 Jul 2016 03:54:02 +0000 (UTC) Received: from mail-oi0-f48.google.com (mail-oi0-f48.google.com [209.85.218.48]) by mx2-lw-eu.apache.org (ASF Mail Server at mx2-lw-eu.apache.org) with ESMTPS id 14B525F47D for ; Fri, 29 Jul 2016 03:54:02 +0000 (UTC) Received: by mail-oi0-f48.google.com with SMTP id w18so90771439oiw.3 for ; Thu, 28 Jul 2016 20:54:02 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20120113; h=from:in-reply-to:references:mime-version:date:message-id:subject:to; bh=9OOUSlRqVRT2VOM8Ge8qkJD+PTnNQ6UrIoASPVkUBXY=; b=LXfM7Ss2YhXOCGgzyd2M1va6gfODQ5inpZWpfNJICB8jx3QpsN5w/9uOFdMfF27Fv8 FCba/TxC1cOp1kRhT1sE/DkY4zQhZu0P0b4QhUCwe0i6UAlyKTPJ6LB26IZlgDh6qh55 TfM9jkqifKNYM74bh3hiOJVCfvYVWfxfjDmDFe2iKvGBuBnrlXdFcPfEGbDpHd8wTsNh G5Y4TucJfNfG834L2w5tiNRw9L3qXSkNiDzkdzM+/nz1lrUagF7+WjDrmXTLc9sNxcNw HFrd4LiezKoDbG+OFvwiK3b4JFEy7xuRcu5C/WucYJya5JSHILlCMqCgd/nroLtvf8SP M+dw== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20130820; h=x-gm-message-state:from:in-reply-to:references:mime-version:date :message-id:subject:to; bh=9OOUSlRqVRT2VOM8Ge8qkJD+PTnNQ6UrIoASPVkUBXY=; b=l05AZ1PTgL2Xq9az2mwVe/Sm5nynLRn6HanL0LLfHVtDkN0iOMHVmrYSWOKOg3dSuf vXtvUxIfh1I1soEqOQT/4bpJA+a/ssmN1KcrtiqVSjQXglodi44pUPd5aHlweP2pbbV9 +dV+c4KbyALeE6yEtbKn280TX6HQc74PgqwRGyoc4+H6MiJqZMws2skcZ2v5HLAyDqFE C8bFV7Jnk3WDI3NZ66u/vrfj5reY9MXHadvw0fBQhrNcLaMxKHu4kap0huXmvWMa8fD2 6ZfxQSWIf0dmy/yPsY3rANy/X0rbEoRg9X4ZOCDBxDDCSjnSxv/nMPDg8dB79ys5CO53 PDVw== X-Gm-Message-State: AEkoouvu5Hd1wdiR1E5NIURKrDWA+MCXUUKSfwbkqZoC43XjbHm1suvFcsHw73uJHlwN/fCBJcGdhvzPruP1FA== X-Received: by 10.157.26.104 with SMTP id u37mr24484457otu.5.1469764434424; Thu, 28 Jul 2016 20:53:54 -0700 (PDT) Received: from 1058052472880 named unknown by gmailapi.google.com with HTTPREST; Thu, 28 Jul 2016 20:53:54 -0700 From: Tai Gordon In-Reply-To: References: X-Mailer: Airmail (367) MIME-Version: 1.0 Date: Thu, 28 Jul 2016 20:53:54 -0700 Message-ID: Subject: Re: Flink kafka group question To: user@flink.apache.org Content-Type: multipart/alternative; boundary=001a1134f9760478700538be33dc archived-at: Fri, 29 Jul 2016 03:54:10 -0000 --001a1134f9760478700538be33dc Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: quoted-printable Hi, 1. What Flink Kafka connector version are you using? 2. How is your non-Flink consumer fetching data from the topic? Is it using the old SimpleConsumer, old High-Level Consumer, or the new consumer API? 3. If you are using the new consumer API, are you using "consumer.assign(=E2=80=A6)" or "consumer.subscribe(=E2=80=A6)"? Internally, the Flink Kafka connectors don=E2=80=99t use the consumer group management functionality because they are using lower-level APIs (SimpleConsumer in 0.8, and KafkaConsumer#assign(=E2=80=A6) in 0.9) on each parallel instance for more control on individual partition consumption. So, essentially, the =E2=80=9Cgroup.id=E2=80=9D setting in the Flink Kafka conn= ector is only used for committing offsets back to ZK / Kafka brokers. From your description I=E2=80=99m not entirely sure of what=E2=80=99s happe= ning, so it=E2=80=99d be helpful if you can provide info about the above questions so we can clarify things further :) Regards, Gordon On July 29, 2016 at 12:11:19 AM, =E7=8E=8B=E8=90=8C (wangm@unicall.cc) wrot= e: Hi, I am using kafka in couple of programs including flink and I am quite confused that how group.id parameter work in flink with kafka comsumer. I have 2 consumers (one in side flink, one outside) running on the same topic and same group.id. From my inpection, they work isolately: If I send one message into the kafka topic, both consumer would receive it. But acording to the kafka document, only one (same topic same group) should be able to get the message. And i tried to start 2 consumers both inside the flink main function, it seems that only one consumer would be receiving all messages from kafka. So, it's quite confused how does flink deal with this kind of situation? Thanks a lot --001a1134f9760478700538be33dc Content-Type: text/html; charset=UTF-8 Content-Transfer-Encoding: quoted-printable =
Hi,

1. What Flink Kafka connector version are you using?=
2. How is your=C2=A0non-Flink= consumer fetching data from the topic? Is it using the old SimpleConsumer,= old High-Level Consumer, or the new consumer API?
3. If you are using the new=C2=A0consumer API, ar= e you using "consumer.assign(=E2=80=A6)" or "consumer.subscr= ibe(=E2=80=A6)"?
Internally, the Flink K= afka connectors don=E2=80=99t use the consumer group management functionali= ty because they are using lower-level APIs (SimpleConsumer in 0.8, and Kafk= aConsumer#assign(=E2=80=A6) in 0.9) on each parallel instance for more cont= rol on individual partition consumption. So, essentially, the=C2=A0=E2=80= =9Cgroup.id=E2=80=9D setting in the Flink K= afka connector is only used for committing offsets back to ZK / Kafka broke= rs.

From your description I=E2=80=99m not ent= irely sure of what=E2=80=99s happening, so it=E2=80=99d be helpful if you c= an provide info about the above questions so we can clarify things further = :)

Regards,
Gordon


On= July 29, 2016 at 12:11:19 AM, =E7=8E=8B=E8=90=8C (wangm@unicall.cc) wrote:

Hi, I am using kafka in couple of programs including flink and I am quite confused that how group.id parameter = work in flink with kafka comsumer.
I have 2 consumers (one in side flink, one outside) running on the same topic and same group.id. From my inpec= tion, they work isolately:
If I send one message into the kafka topic, both consumer would receive it. But acording to the kafka document, only one (same topic same group) should be able to get the message.

And i tried to start 2 consumers both inside the flink main function, it seems that only one consumer would be receiving all messages from kafka.

So, it's quite confused how does flink deal with this kind of situation?

Thanks a lot

--001a1134f9760478700538be33dc--