<gh_stars>0
#!/usr/bin/env python
#
# Copyright 2016 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

#
# Example high-level Kafka 0.9 balanced Consumer
#
import pickle
import sys
import getopt
import json
import logging
from contextlib import closing

from confluent_kafka import Consumer, KafkaException
from pprint import pformat


def stats_cb(stats_json_str):
    stats_json = json.loads(stats_json_str)
    print('\nKAFKA Stats: {}\n'.format(pformat(stats_json)))


def print_usage_and_exit(program_name):
    sys.stderr.write('Usage: %s [options..] <bootstrap-brokers> <group> <topic1> <topic2> ..\n' % program_name)
    options = '''
 Options:
  -T <intvl>   Enable client statistics at specified interval (ms)
'''
    sys.stderr.write(options)
    sys.exit(1)


if __name__ == '__main__':
    optlist, argv = getopt.getopt(sys.argv[1:], 'T:')
    if len(argv) < 3:
        print_usage_and_exit(sys.argv[0])

    broker = argv[0]
    group = argv[1]
    topics = argv[2:]

    # Consumer configuration
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    conf = {'bootstrap.servers': broker,
            'group.id': group,
            'session.timeout.ms': 6000,
            'auto.offset.reset': 'earliest'}


    # Check to see if -T option exists
    for opt in optlist:
        if opt[0] != '-T':
            continue
        try:
            intval = int(opt[1])
        except ValueError:
            sys.stderr.write("Invalid option value for -T: %s\n" % opt[1])
            sys.exit(1)

        if intval <= 0:
            sys.stderr.write("-T option value needs to be larger than zero: %s\n" % opt[1])
            sys.exit(1)

        conf['stats_cb'] = stats_cb
        conf['statistics.interval.ms'] = int(opt[1])

    # Create logger for consumer (logs will be emitted when poll() is called)
    logger = logging.getLogger('consumer')
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(asctime)-15s %(levelname)-8s %(message)s'))
    logger.addHandler(handler)

    # # Create Consumer instance
    # # Hint: try debug='fetch' to generate some log messages
    # consumer = Consumer(conf, logger=logger)
    #
    # def print_assignment(consumer, partitions):
    #     print('Assignment:', partitions)
    #
    # # Subscribe to topics
    # consumer.subscribe(topics, on_assign=print_assignment)
    #
    # # Read messages from Kafka, print to stdout
    # try:
    #     while True:
    #         msg = consumer.poll(timeout=1.0)
    #         if msg is None:
    #             continue
    #         if msg.error():
    #             raise KafkaException(msg.error())
    #
    #         # Proper message
    #         # sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' %
    #         #                  (msg.topic(), msg.partition(), msg.offset(),
    #         #                   str(msg.key())))
    #         # print(msg.value())
    #         data = pickle.loads(msg.value())
    #         print(f'{data=}')
    #
    # except KeyboardInterrupt:
    #     sys.stderr.write('%% Aborted by user\n')
    #
    # finally:
    #     # Close down consumer to commit final offsets.
    #     consumer.close()


    def set_offset_to_last_unread_message(consumer, partitions):
        logger.debug('Subscription assigned')
        logger.debug(f'{type(partitions)} {partitions=}')

        partitions_committed = consumer.committed(partitions)
        logger.debug(f'{partitions_committed=}')

        for i in range(len(partitions)):
            committed_offset = partitions_committed[i].offset
            logger.debug(f'{committed_offset=}')
            first_offset_index, end_offset_index = consumer.get_watermark_offsets(partitions[i])
            logger.debug(f'{first_offset_index=} {end_offset_index=}')
            last_offset = end_offset_index - 1
            logger.debug(f'{last_offset=}')
            if committed_offset < last_offset:
                logger.debug(f'Set offset to the last message')
                partitions[i].offset = last_offset

        consumer.assign(partitions)
        logger.debug('Partitions reassigned')

    with closing(Consumer(conf, logger=logger)) as consumer:
        # Подписаться на топик
        consumer.subscribe(topics, on_assign=set_offset_to_last_unread_message)

        try:
            while True:
                msg = consumer.poll(timeout=1.0)
                if msg is None:
                    continue
                if msg.error():
                    raise KafkaException(msg.error())

                # Proper message
                # sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' %
                #                  (msg.topic(), msg.partition(), msg.offset(),
                #                   str(msg.key())))
                # print(msg.value())
                data = pickle.loads(msg.value())
                print(f'{msg.headers()=}')
                print(f'{data=}')

        except KeyboardInterrupt:
            sys.stderr.write('%% Aborted by user\n')



"""
var consumer = new ConsumerBuilder<Null, string>(config)
                .SetPartitionsAssignedHandler((c, tps) =>
                {
                    var partitionOffsets = c.Committed(tps, TimeSpan.FromSeconds(10));
                    var watermarkOffsets = tps.Select(tp => c.QueryWatermarkOffsets(tp, TimeSpan.FromSeconds(10)));
                    var offsets = watermarkOffsets.Zip(partitionOffsets, (watermarkOffset, topicPartitionOffset) =>
                    {
                        if (topicPartitionOffset.Offset.IsSpecial || watermarkOffset.High.IsSpecial)
                        {
                            return topicPartitionOffset.Offset;
                        }

                        var lastTopicOffset = watermarkOffset.High - 1;
                        var lastCommittedOffset = topicPartitionOffset.Offset - 1;

                        if (lastTopicOffset == 0)
                        {
                            return new Offset(0);
                        }
                        if (lastCommittedOffset == lastTopicOffset)
                        {
                            return new Offset(lastCommittedOffset);
                        }
                        return new Offset(lastCommittedOffset + 1);
                    });

                    return tps.Zip(offsets, (partition, offset) => new TopicPartitionOffset(partition, offset));
                })
                .Build()
"""
#!/usr/bin/env python
#
# Copyright 2016 Confluent Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

#
# Example high-level Kafka 0.9 balanced Consumer
#
import pickle
import sys
import getopt
import json
import logging
from contextlib import closing

from confluent_kafka import Consumer, KafkaException
from pprint import pformat


def stats_cb(stats_json_str):
    stats_json = json.loads(stats_json_str)
    print('\nKAFKA Stats: {}\n'.format(pformat(stats_json)))


def print_usage_and_exit(program_name):
    sys.stderr.write('Usage: %s [options..] <bootstrap-brokers> <group> <topic1> <topic2> ..\n' % program_name)
    options = '''
 Options:
  -T <intvl>   Enable client statistics at specified interval (ms)
'''
    sys.stderr.write(options)
    sys.exit(1)


if __name__ == '__main__':
    optlist, argv = getopt.getopt(sys.argv[1:], 'T:')
    if len(argv) < 3:
        print_usage_and_exit(sys.argv[0])

    broker = argv[0]
    group = argv[1]
    topics = argv[2:]

    # Consumer configuration
    # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
    conf = {'bootstrap.servers': broker,
            'group.id': group,
            'session.timeout.ms': 6000,
            'auto.offset.reset': 'earliest'}


    # Check to see if -T option exists
    for opt in optlist:
        if opt[0] != '-T':
            continue
        try:
            intval = int(opt[1])
        except ValueError:
            sys.stderr.write("Invalid option value for -T: %s\n" % opt[1])
            sys.exit(1)

        if intval <= 0:
            sys.stderr.write("-T option value needs to be larger than zero: %s\n" % opt[1])
            sys.exit(1)

        conf['stats_cb'] = stats_cb
        conf['statistics.interval.ms'] = int(opt[1])

    # Create logger for consumer (logs will be emitted when poll() is called)
    logger = logging.getLogger('consumer')
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(asctime)-15s %(levelname)-8s %(message)s'))
    logger.addHandler(handler)

    # # Create Consumer instance
    # # Hint: try debug='fetch' to generate some log messages
    # consumer = Consumer(conf, logger=logger)
    #
    # def print_assignment(consumer, partitions):
    #     print('Assignment:', partitions)
    #
    # # Subscribe to topics
    # consumer.subscribe(topics, on_assign=print_assignment)
    #
    # # Read messages from Kafka, print to stdout
    # try:
    #     while True:
    #         msg = consumer.poll(timeout=1.0)
    #         if msg is None:
    #             continue
    #         if msg.error():
    #             raise KafkaException(msg.error())
    #
    #         # Proper message
    #         # sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' %
    #         #                  (msg.topic(), msg.partition(), msg.offset(),
    #         #                   str(msg.key())))
    #         # print(msg.value())
    #         data = pickle.loads(msg.value())
    #         print(f'{data=}')
    #
    # except KeyboardInterrupt:
    #     sys.stderr.write('%% Aborted by user\n')
    #
    # finally:
    #     # Close down consumer to commit final offsets.
    #     consumer.close()


    def set_offset_to_last_unread_message(consumer, partitions):
        logger.debug('Subscription assigned')
        logger.debug(f'{type(partitions)} {partitions=}')

        partitions_committed = consumer.committed(partitions)
        logger.debug(f'{partitions_committed=}')

        for i in range(len(partitions)):
            committed_offset = partitions_committed[i].offset
            logger.debug(f'{committed_offset=}')
            first_offset_index, end_offset_index = consumer.get_watermark_offsets(partitions[i])
            logger.debug(f'{first_offset_index=} {end_offset_index=}')
            last_offset = end_offset_index - 1
            logger.debug(f'{last_offset=}')
            if committed_offset < last_offset:
                logger.debug(f'Set offset to the last message')
                partitions[i].offset = last_offset

        consumer.assign(partitions)
        logger.debug('Partitions reassigned')

    with closing(Consumer(conf, logger=logger)) as consumer:
        # Подписаться на топик
        consumer.subscribe(topics, on_assign=set_offset_to_last_unread_message)

        try:
            while True:
                msg = consumer.poll(timeout=1.0)
                if msg is None:
                    continue
                if msg.error():
                    raise KafkaException(msg.error())

                # Proper message
                # sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' %
                #                  (msg.topic(), msg.partition(), msg.offset(),
                #                   str(msg.key())))
                # print(msg.value())
                data = pickle.loads(msg.value())
                print(f'{msg.headers()=}')
                print(f'{data=}')

        except KeyboardInterrupt:
            sys.stderr.write('%% Aborted by user\n')



"""
var consumer = new ConsumerBuilder<Null, string>(config)
                .SetPartitionsAssignedHandler((c, tps) =>
                {
                    var partitionOffsets = c.Committed(tps, TimeSpan.FromSeconds(10));
                    var watermarkOffsets = tps.Select(tp => c.QueryWatermarkOffsets(tp, TimeSpan.FromSeconds(10)));
                    var offsets = watermarkOffsets.Zip(partitionOffsets, (watermarkOffset, topicPartitionOffset) =>
                    {
                        if (topicPartitionOffset.Offset.IsSpecial || watermarkOffset.High.IsSpecial)
                        {
                            return topicPartitionOffset.Offset;
                        }

                        var lastTopicOffset = watermarkOffset.High - 1;
                        var lastCommittedOffset = topicPartitionOffset.Offset - 1;

                        if (lastTopicOffset == 0)
                        {
                            return new Offset(0);
                        }
                        if (lastCommittedOffset == lastTopicOffset)
                        {
                            return new Offset(lastCommittedOffset);
                        }
                        return new Offset(lastCommittedOffset + 1);
                    });

                    return tps.Zip(offsets, (partition, offset) => new TopicPartitionOffset(partition, offset));
                })
                .Build()
"""
#!/usr/bin/env python # # Copyright 2016 Confluent Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # # Example high-level Kafka 0.9 balanced Consumer # Options:   -T <intvl>   Enable client statistics at specified interval (ms) # Consumer configuration # See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md # Check to see if -T option exists # Create logger for consumer (logs will be emitted when poll() is called) # # Create Consumer instance # # Hint: try debug='fetch' to generate some log messages # consumer = Consumer(conf, logger=logger) # # def print_assignment(consumer, partitions): #     print('Assignment:', partitions) # # # Subscribe to topics # consumer.subscribe(topics, on_assign=print_assignment) # # # Read messages from Kafka, print to stdout # try: #     while True: #         msg = consumer.poll(timeout=1.0) #         if msg is None: #             continue #         if msg.error(): #             raise KafkaException(msg.error()) # #         # Proper message #         # sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' % #         #                  (msg.topic(), msg.partition(), msg.offset(), #         #                   str(msg.key()))) #         # print(msg.value()) #         data = pickle.loads(msg.value()) #         print(f'{data=}') # # except KeyboardInterrupt: #     sys.stderr.write('%% Aborted by user\n') # # finally: #     # Close down consumer to commit final offsets. #     consumer.close() # Подписаться на топик # Proper message # sys.stderr.write('%% %s [%d] at offset %d with key %s:\n' % #                  (msg.topic(), msg.partition(), msg.offset(), #                   str(msg.key()))) # print(msg.value()) var consumer = new ConsumerBuilder<Null, string>(config)                 .SetPartitionsAssignedHandler((c, tps) =>                 {                     var partitionOffsets = c.Committed(tps, TimeSpan.FromSeconds(10));                     var watermarkOffsets = tps.Select(tp => c.QueryWatermarkOffsets(tp, TimeSpan.FromSeconds(10)));                     var offsets = watermarkOffsets.Zip(partitionOffsets, (watermarkOffset, topicPartitionOffset) =>                     {                         if (topicPartitionOffset.Offset.IsSpecial || watermarkOffset.High.IsSpecial)                         {                             return topicPartitionOffset.Offset;                         }                          var lastTopicOffset = watermarkOffset.High - 1;                         var lastCommittedOffset = topicPartitionOffset.Offset - 1;                          if (lastTopicOffset == 0)                         {                             return new Offset(0);                         }                         if (lastCommittedOffset == lastTopicOffset)                         {                             return new Offset(lastCommittedOffset);                         }                         return new Offset(lastCommittedOffset + 1);                     });                      return tps.Zip(offsets, (partition, offset) => new TopicPartitionOffset(partition, offset));                 })                 .Build()