src/main/java/com/mapr/fuse/client/TopicReader.java
public Stream<Bytes> readPartition(final TopicPartition partition, final long offset,
                                       final long amount, final long timeout) {
        final AtomicBoolean closed = new AtomicBoolean(false);
        long currentPosition = offset;
        List<ConsumerRecord<Bytes, Bytes>> records = new LinkedList<>();
        Stopwatch stopwatch = Stopwatch.createStarted();
        kafkaConsumer.assign(Collections.singletonList(partition));
        while (!closed.get()) {
            try {
                if (kafkaConsumer.assignment().isEmpty()) {
                    Thread.sleep(100);
                    if (stopwatch.elapsed(TimeUnit.MILLISECONDS) >= timeout) {
                        closed.set(true);
                    } else continue;
                }

                kafkaConsumer.seek(partition, currentPosition);

                ConsumerRecords<Bytes, Bytes> consumerRecords = kafkaConsumer.poll(500);
                if (Objects.isNull(consumerRecords) || consumerRecords.isEmpty()) {
                    if (stopwatch.elapsed(TimeUnit.MILLISECONDS) >= timeout) {
                        closed.set(true);
                    } else continue;
                }
                records.addAll(consumerRecords.records(partition));
                currentPosition += consumerRecords.count();

                if (records.size() >= amount || stopwatch.elapsed(TimeUnit.MILLISECONDS) >= timeout) {
                    closed.set(true);
                }
            } catch (UnknownTopicOrPartitionException ignored) {
            } catch (Exception e) {
                log.error("Unexpected error: {}", e.getMessage(), e);
            }
        }
        kafkaConsumer.unsubscribe();
        return records.stream().map(ConsumerRecord::value);
    }
/**
     * Used to read needed amount of messages from topic. If it is not possible to read (less messages available) timeout
     * will brake loop
     *
     * @param partition partition of topic to read from
     * @param offset    initial offset
     * @param amount    amount of messages to read
     * @param timeout   timeout to brake the loop if it is not possible te read needed amount of messages
     * @return List of records
     */
Used to read needed amount of messages from topic. If it is not possible to read (less messages available) timeout
will brake loop


Used to read needed amount of messages from topic . If it is not possible to read ( less messages available ) timeout will brake loop
public Stream<Bytes> readPartition(final TopicPartition partition, final long offset,
                                       final long amount, final long timeout) {
        final AtomicBoolean closed = new AtomicBoolean(false);
        long currentPosition = offset;
        List<ConsumerRecord<Bytes, Bytes>> records = new LinkedList<>();
        Stopwatch stopwatch = Stopwatch.createStarted();
        kafkaConsumer.assign(Collections.singletonList(partition));
        while (!closed.get()) {
            try {
                if (kafkaConsumer.assignment().isEmpty()) {
                    Thread.sleep(100);
                    if (stopwatch.elapsed(TimeUnit.MILLISECONDS) >= timeout) {
                        closed.set(true);
                    } else continue;
                }
                kafkaConsumer.seek(partition, currentPosition);
                ConsumerRecords<Bytes, Bytes> consumerRecords = kafkaConsumer.poll(500);
                if (Objects.isNull(consumerRecords) || consumerRecords.isEmpty()) {
                    if (stopwatch.elapsed(TimeUnit.MILLISECONDS) >= timeout) {
                        closed.set(true);
                    } else continue;
                }
                records.addAll(consumerRecords.records(partition));
                currentPosition += consumerRecords.count();
                if (records.size() >= amount || stopwatch.elapsed(TimeUnit.MILLISECONDS) >= timeout) {
                    closed.set(true);
                }
            } catch (UnknownTopicOrPartitionException ignored) {
            } catch (Exception e) {
                log.error("Unexpected error: {}", e.getMessage(), e);
            }
        }
        kafkaConsumer.unsubscribe();
        return records.stream().map(ConsumerRecord::value);
    }
public Stream < Bytes > readPartition ( final TopicPartition partition , final long offset , final long amount , final long timeout ) { final AtomicBoolean closed = new AtomicBoolean ( false ) ; long currentPosition = offset ; List < ConsumerRecord < Bytes , Bytes > > records = new LinkedList < > ( ) ; Stopwatch stopwatch = Stopwatch . createStarted ( ) ; kafkaConsumer . assign ( Collections . singletonList ( partition ) ) ; while ( ! closed . get ( ) ) { try { if ( kafkaConsumer . assignment ( ) . isEmpty ( ) ) { Thread . sleep ( 100 ) ; if ( stopwatch . elapsed ( TimeUnit . MILLISECONDS ) >= timeout ) { closed . set ( true ) ; } else continue ; } kafkaConsumer . seek ( partition , currentPosition ) ; ConsumerRecords < Bytes , Bytes > consumerRecords = kafkaConsumer . poll ( 500 ) ; if ( Objects . isNull ( consumerRecords ) || consumerRecords . isEmpty ( ) ) { if ( stopwatch . elapsed ( TimeUnit . MILLISECONDS ) >= timeout ) { closed . set ( true ) ; } else continue ; } records . addAll ( consumerRecords . records ( partition ) ) ; currentPosition += consumerRecords . count ( ) ; if ( records . size ( ) >= amount || stopwatch . elapsed ( TimeUnit . MILLISECONDS ) >= timeout ) { closed . set ( true ) ; } } catch ( UnknownTopicOrPartitionException ignored ) { } catch ( Exception e ) { log . error ( "Unexpected error: {}" , e . getMessage ( ) , e ) ; } } kafkaConsumer . unsubscribe ( ) ; return records . stream ( ) . map ( ConsumerRecord :: value ) ; }
Used to read needed amount of messages from topic.
Used to read needed amount of messages from topic .