flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Robert Metzger <rmetz...@apache.org>
Subject Re: Issue Faced: Not able to get the consumer offsets from Kafka when using Flink with Flink-Kafka Connector
Date Tue, 07 Feb 2017 17:38:53 GMT
Hi Mahesh,

this is a known limitation of Apache Kafka:
https://www.mail-archive.com/users@kafka.apache.org/msg22595.html
You could implement a tool that is manually retrieving the latest offset
for the group from the __offsets topic.

On Tue, Feb 7, 2017 at 6:24 PM, MAHESH KUMAR <r.mahesh.kumar.blr@gmail.com>
wrote:

> Hi Team,
>
> Kindly let me know if I am doing something wrong.
>
> Kafka Version - kafka_2.11-0.10.1.1
> Flink Version - flink-1.2.0
> Using the latest Kafka Connector - FlinkKafkaConsumer010 -
> flink-connector-kafka-0.10_2.11
>
> Issue Faced: Not able to get the consumer offsets from Kafka when using
> Flink with Flink-Kafka Connector
>
> $ /work/kafka_2.11-0.10.1.1/bin/kafka-consumer-groups.sh
> --bootstrap-server localhost:9092  --list
> console-consumer-19886
> console-consumer-89637
> $
>
> It does not show the consumer group "test"
>
> For a group that does not exist, the message is as follows:
>
> $ /work/kafka_2.11-0.10.1.1/bin/kafka-consumer-groups.sh
> --bootstrap-server localhost:9092 --group test1 --describe
> Consumer group `test1` does not exist.
> $
>
> For the "test" group the error message is as follows
>
> $ /work/kafka_2.11-0.10.1.1/bin/kafka-consumer-groups.sh
> --bootstrap-server localhost:9092 --group test --describe
> Error while executing consumer group command Group test with protocol type
> '' is not a valid consumer group
> java.lang.IllegalArgumentException: Group test with protocol type '' is
> not a valid consumer group
> at kafka.admin.AdminClient.describeConsumerGroup(AdminClient.scala:152)
> at kafka.admin.ConsumerGroupCommand$KafkaConsumerGroupService.d
> escribeGroup(ConsumerGroupCommand.scala:308)
> at kafka.admin.ConsumerGroupCommand$ConsumerGroupService$class.
> describe(ConsumerGroupCommand.scala:89)
> at kafka.admin.ConsumerGroupCommand$KafkaConsumerGroupService.d
> escribe(ConsumerGroupCommand.scala:296)
> at kafka.admin.ConsumerGroupCommand$.main(ConsumerGroupCommand.scala:68)
> at kafka.admin.ConsumerGroupCommand.main(ConsumerGroupCommand.scala)
> $
>
> The error is from the AdminClient.scala (https://github.com/apache/kaf
> ka/blob/trunk/core/src/main/scala/kafka/admin/AdminClient.scala)
>
>     if (metadata.state != "Dead" && metadata.state != "Empty" &&
> metadata.protocolType != ConsumerProtocol.PROTOCOL_TYPE)
>       throw new IllegalArgumentException(s"Consumer Group $groupId with
> protocol type '${metadata.protocolType}' is not a valid consumer group")
>
> Code:
>
> import java.util.Properties;
> import org.apache.flink.streaming.api.datastream.DataStreamSource;
> import org.apache.flink.streaming.api.environment.StreamExecutionEn
> vironment;
> import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
> import org.apache.flink.streaming.util.serialization.SimpleStringSchema;
>
> public class KafkaFlinkOutput {
>     private static final String LOCAL_ZOOKEEPER_HOST = "localhost:2181";
>     private static final String LOCAL_KAFKA_BROKER = "localhost:9092";
>     private static final String CONSUMER_GROUP = "test";
>
>     public KafkaFlinkOutput() {
>     }
>
>     public static void main(String[] args) throws Exception {
>         StreamExecutionEnvironment env = StreamExecutionEnvironment.get
> ExecutionEnvironment();
>         Properties kafkaProps = new Properties();
>         kafkaProps.setProperty("zookeeper.connect", "localhost:2181");
>         kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
>         kafkaProps.setProperty("group.id", "test");
>         kafkaProps.setProperty("auto.offset.reset", "latest");
>         env.enableCheckpointing(1000L);
>         FlinkKafkaConsumer010 consumer = new FlinkKafkaConsumer010("testIn1",
> new SimpleStringSchema(), kafkaProps);
>         DataStreamSource consumerData = env.addSource(consumer);
>         consumerData.print();
>         System.out.println("Streaming Kafka in Flink");
>         env.execute("Starting now!");
>     }
> }
>
> Debug Logs that show that Kafka Connector does commit to Kafka:
>
> 2017-02-07 09:52:38,851 INFO  org.apache.kafka.clients.consumer.ConsumerConfig
>              - ConsumerConfig values:
>         metric.reporters = []
>         metadata.max.age.ms = 300000
>         partition.assignment.strategy = [org.apache.kafka.clients.cons
> umer.RangeAssignor]
>         reconnect.backoff.ms = 50
>         sasl.kerberos.ticket.renew.window.factor = 0.8
>         max.partition.fetch.bytes = 1048576
>         bootstrap.servers = [localhost:9092]
>         ssl.keystore.type = JKS
>         enable.auto.commit = true
>         sasl.mechanism = GSSAPI
>         interceptor.classes = null
>         exclude.internal.topics = true
>         ssl.truststore.password = null
>         client.id =
>         ssl.endpoint.identification.algorithm = null
>         max.poll.records = 2147483647 <02147%20483%20647>
>         check.crcs = true
>         request.timeout.ms = 40000
>         heartbeat.interval.ms = 3000
>         auto.commit.interval.ms = 5000
>         receive.buffer.bytes = 65536
>         ssl.truststore.type = JKS
>         ssl.truststore.location = null
>         ssl.keystore.password = null
>         fetch.min.bytes = 1
>         send.buffer.bytes = 131072
>         value.deserializer = class org.apache.kafka.common.serial
> ization.ByteArrayDeserializer
>         group.id = test
>         retry.backoff.ms = 100
>         sasl.kerberos.kinit.cmd = /usr/bin/kinit
>         sasl.kerberos.service.name = null
>         sasl.kerberos.ticket.renew.jitter = 0.05
>         ssl.trustmanager.algorithm = PKIX
>         ssl.key.password = null
>         fetch.max.wait.ms = 500
>         sasl.kerberos.min.time.before.relogin = 60000
>         connections.max.idle.ms = 540000
>         session.timeout.ms = 30000
>         metrics.num.samples = 2
>         key.deserializer = class org.apache.kafka.common.serial
> ization.ByteArrayDeserializer
>         ssl.protocol = TLS
>         ssl.provider = null
>         ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
>         ssl.keystore.location = null
>         ssl.cipher.suites = null
>         security.protocol = PLAINTEXT
>         ssl.keymanager.algorithm = SunX509
>         metrics.sample.window.ms = 30000
>         auto.offset.reset = latest
>
>
> 2017-02-07 09:53:38,524 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Sending heartbeat to JobManager
> 2017-02-07 09:53:38,731 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Triggering checkpoint 12 @ 1486486418731
> 2017-02-07 09:53:38,731 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver TriggerCheckpoint 12@1486486418731 for
> cfe59bc4aadc96e5a2235581460e9f3d.
> 2017-02-07 09:53:38,731 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Trigger for Source:
> Custom Source -> Sink: Unnamed (1/4) (cfe59bc4aadc96e5a2235581460e9f3d).
> on task Source: Custom Source -> Sink: Unnamed (1/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver TriggerCheckpoint 12@1486486418731 for
> 5611d20817d9a49680117c9ab000116c.
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Trigger for Source:
> Custom Source -> Sink: Unnamed (2/4) (5611d20817d9a49680117c9ab000116c).
> on task Source: Custom Source -> Sink: Unnamed (2/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver TriggerCheckpoint 12@1486486418731 for
> 58e5bc1040fc99f8d3f0a32c2bd524b6.
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Trigger for Source:
> Custom Source -> Sink: Unnamed (3/4) (58e5bc1040fc99f8d3f0a32c2bd524b6).
> on task Source: Custom Source -> Sink: Unnamed (3/4)
> 2017-02-07 09:53:38,732 INFO  org.apache.flink.core.fs.FileSystem
>                   - Created new CloseableRegistry
> org.apache.flink.core.fs.SafetyNetCloseableRegistry@26590268 for Async
> calls on Source: Custom Source -> Sink: Unnamed (1/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Starting checkpoint 12 on task Source: Custom Source -> Sink:
> Unnamed (1/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver TriggerCheckpoint 12@1486486418731 for
> 95ad7256919d3296b37d17693cd0ba71.
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Trigger for Source:
> Custom Source -> Sink: Unnamed (4/4) (95ad7256919d3296b37d17693cd0ba71).
> on task Source: Custom Source -> Sink: Unnamed (4/4)
> 2017-02-07 09:53:38,732 INFO  org.apache.flink.core.fs.FileSystem
>                   - Created new CloseableRegistry
> org.apache.flink.core.fs.SafetyNetCloseableRegistry@2b3c1766 for Async
> calls on Source: Custom Source -> Sink: Unnamed (2/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Starting checkpoint 12 on task Source: Custom Source -> Sink:
> Unnamed (2/4)
> 2017-02-07 09:53:38,732 INFO  org.apache.flink.core.fs.FileSystem
>                   - Created new CloseableRegistry
> org.apache.flink.core.fs.SafetyNetCloseableRegistry@2c4eb75a for Async
> calls on Source: Custom Source -> Sink: Unnamed (4/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Starting checkpoint 12 on task Source: Custom Source -> Sink:
> Unnamed (4/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Finished synchronous checkpoints for checkpoint 12 on task
> Source: Custom Source -> Sink: Unnamed (2/4)
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (2/4) - finished
> synchronous part of checkpoint 12.Alignment duration: 0 ms, snapshot
> duration 0 ms
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Finished synchronous checkpoints for checkpoint 12 on task
> Source: Custom Source -> Sink: Unnamed (1/4)
> 2017-02-07 09:53:38,732 INFO  org.apache.flink.core.fs.FileSystem
>                   - Ensuring all FileSystem streams are closed for Async
> calls on Source: Custom Source -> Sink: Unnamed (2/4)
> 2017-02-07 09:53:38,732 INFO  org.apache.flink.core.fs.FileSystem
>                   - Created new CloseableRegistry
> org.apache.flink.core.fs.SafetyNetCloseableRegistry@b0b9bf0 for Async
> calls on Source: Custom Source -> Sink: Unnamed (3/4)
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Starting checkpoint 12 on task Source: Custom Source -> Sink:
> Unnamed (3/4)
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (1/4) - finished
> asynchronous part of checkpoint 12. Asynchronous duration: 0 ms
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Finished synchronous checkpoints for checkpoint 12 on task
> Source: Custom Source -> Sink: Unnamed (4/4)
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Finished synchronous checkpoints for checkpoint 12 on task
> Source: Custom Source -> Sink: Unnamed (3/4)
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (3/4) - finished
> synchronous part of checkpoint 12.Alignment duration: 0 ms, snapshot
> duration 0 ms
> 2017-02-07 09:53:38,733 INFO  org.apache.flink.core.fs.FileSystem
>                   - Ensuring all FileSystem streams are closed for Async
> calls on Source: Custom Source -> Sink: Unnamed (3/4)
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (2/4) - finished
> asynchronous part of checkpoint 12. Asynchronous duration: 0 ms
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (3/4) - finished
> asynchronous part of checkpoint 12. Asynchronous duration: 0 ms
> 2017-02-07 09:53:38,732 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (1/4) - finished
> synchronous part of checkpoint 12.Alignment duration: 0 ms, snapshot
> duration 0 ms
> 2017-02-07 09:53:38,733 INFO  org.apache.flink.core.fs.FileSystem
>                   - Ensuring all FileSystem streams are closed for Async
> calls on Source: Custom Source -> Sink: Unnamed (1/4)
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (4/4) - finished
> asynchronous part of checkpoint 12. Asynchronous duration: 0 ms
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Received acknowledge message for checkpoint 12 from task
> 5611d20817d9a49680117c9ab000116c of job ef5998a3d1cbba24cb1790564f3037bf.
> 2017-02-07 09:53:38,733 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Source: Custom Source -> Sink: Unnamed (4/4) - finished
> synchronous part of checkpoint 12.Alignment duration: 0 ms, snapshot
> duration 0 ms
> 2017-02-07 09:53:38,734 INFO  org.apache.flink.core.fs.FileSystem
>                   - Ensuring all FileSystem streams are closed for Async
> calls on Source: Custom Source -> Sink: Unnamed (4/4)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Received acknowledge message for checkpoint 12 from task
> 58e5bc1040fc99f8d3f0a32c2bd524b6 of job ef5998a3d1cbba24cb1790564f3037bf.
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Received acknowledge message for checkpoint 12 from task
> 95ad7256919d3296b37d17693cd0ba71 of job ef5998a3d1cbba24cb1790564f3037bf.
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Received acknowledge message for checkpoint 12 from task
> cfe59bc4aadc96e5a2235581460e9f3d of job ef5998a3d1cbba24cb1790564f3037bf.
> 2017-02-07 09:53:38,734 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Completed checkpoint 12 (4490 bytes in 3 ms).
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Checkpoint state: TaskState(jobVertexID:
> cbc357ccb763df2852fee8c4fc7d55f2, parallelism: 4, sub task states: 4,
> total size (bytes): 4490)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver ConfirmCheckpoint 12@1486486418731 for
> cfe59bc4aadc96e5a2235581460e9f3d.
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Confirmation for Source:
> Custom Source -> Sink: Unnamed (1/4) on task Source: Custom Source -> Sink:
> Unnamed (1/4)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver ConfirmCheckpoint 12@1486486418731 for
> 5611d20817d9a49680117c9ab000116c.
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Notification of complete checkpoint for task Source: Custom
> Source -> Sink: Unnamed (1/4)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Confirmation for Source:
> Custom Source -> Sink: Unnamed (2/4) on task Source: Custom Source -> Sink:
> Unnamed (2/4)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.streaming.con
> nectors.kafka.FlinkKafkaConsumerBase  - Committing offsets to
> Kafka/ZooKeeper for checkpoint 12
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver ConfirmCheckpoint 12@1486486418731 for
> 58e5bc1040fc99f8d3f0a32c2bd524b6.
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Confirmation for Source:
> Custom Source -> Sink: Unnamed (3/4) on task Source: Custom Source -> Sink:
> Unnamed (3/4)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Notification of complete checkpoint for task Source: Custom
> Source -> Sink: Unnamed (2/4)
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.streaming.con
> nectors.kafka.FlinkKafkaConsumerBase  - Committing offsets to
> Kafka/ZooKeeper for checkpoint 12
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Notification of complete checkpoint for task Source: Custom
> Source -> Sink: Unnamed (3/4)
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.con
> nectors.kafka.FlinkKafkaConsumerBase  - Committing offsets to
> Kafka/ZooKeeper for checkpoint 12
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.con
> nectors.kafka.internal.Kafka09Fetcher  - Sending async offset commit
> request to Kafka broker
> 2017-02-07 09:53:38,734 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Receiver ConfirmCheckpoint 12@1486486418731 for
> 95ad7256919d3296b37d17693cd0ba71.
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.runtime.taskmanager.Task
>                   - Invoking async call Checkpoint Confirmation for Source:
> Custom Source -> Sink: Unnamed (4/4) on task Source: Custom Source -> Sink:
> Unnamed (4/4)
> 2017-02-07 09:53:38,735 DEBUG org.apache.kafka.clients.consumer.KafkaConsumer
>               - Committing offsets: {testIn1-1=OffsetAndMetadata{offset=17421,
> metadata=''}, testIn1-5=OffsetAndMetadata{offset=0, metadata=''},
> testIn1-9=OffsetAndMetadata{offset=17493, metadata=''}}
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.runtime.tasks.StreamTask
>           - Notification of complete checkpoint for task Source: Custom
> Source -> Sink: Unnamed (4/4)
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.con
> nectors.kafka.FlinkKafkaConsumerBase  - Committing offsets to
> Kafka/ZooKeeper for checkpoint 12
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.con
> nectors.kafka.internal.Kafka09Fetcher  - Sending async offset commit
> request to Kafka broker
> 2017-02-07 09:53:38,735 DEBUG org.apache.kafka.clients.consumer.KafkaConsumer
>               - Committing offsets: {testIn1-0=OffsetAndMetadata{offset=34926,
> metadata=''}, testIn1-4=OffsetAndMetadata{offset=17325, metadata=''},
> testIn1-8=OffsetAndMetadata{offset=17564, metadata=''}}
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.con
> nectors.kafka.internal.Kafka09Fetcher  - Sending async offset commit
> request to Kafka broker
> 2017-02-07 09:53:38,736 DEBUG org.apache.kafka.clients.consumer.KafkaConsumer
>               - Committing offsets: {testIn1-3=OffsetAndMetadata{offset=52292,
> metadata=''}, testIn1-7=OffsetAndMetadata{offset=0, metadata=''}}
> 2017-02-07 09:53:38,735 DEBUG org.apache.flink.streaming.con
> nectors.kafka.internal.Kafka09Fetcher  - Sending async offset commit
> request to Kafka broker
> 2017-02-07 09:53:38,736 DEBUG org.apache.kafka.clients.consumer.KafkaConsumer
>               - Committing offsets: {testIn1-2=OffsetAndMetadata{offset=0,
> metadata=''}, testIn1-6=OffsetAndMetadata{offset=17438, metadata=''}}
> 2017-02-07 09:53:38,737 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 0 for
> partition testIn1-5
> 2017-02-07 09:53:38,737 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 17493
> for partition testIn1-9
> 2017-02-07 09:53:38,737 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 17421
> for partition testIn1-1
> 2017-02-07 09:53:38,737 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 17325
> for partition testIn1-4
> 2017-02-07 09:53:38,738 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 17564
> for partition testIn1-8
> 2017-02-07 09:53:38,738 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 34926
> for partition testIn1-0
> 2017-02-07 09:53:38,738 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 0 for
> partition testIn1-7
> 2017-02-07 09:53:38,738 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 52292
> for partition testIn1-3
> 2017-02-07 09:53:38,738 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 17438
> for partition testIn1-6
> 2017-02-07 09:53:38,738 DEBUG org.apache.kafka.clients.consu
> mer.internals.ConsumerCoordinator  - Group test committed offset 0 for
> partition testIn1-2
> 2017-02-07 09:53:43,524 DEBUG org.apache.flink.runtime.taskmanager.TaskManager
>              - Sending heartbeat to JobManager
> 2017-02-07 09:53:43,730 INFO  org.apache.flink.runtime.checkpoint.CheckpointCoordinator
>     - Triggering checkpoint 13 @ 1486486423730
>
> --
>
> Mahesh Kumar Ravindranathan
> Data Streaming Engineer
> Oracle Marketing Cloud - Social Platform
> Contact No:+1(720)492-4445
>
>

Mime
View raw message