Skip to content

Commit 48d93af

Browse files
author
accelerated
committed
added test case for polling strategy refactored the strategy class
1 parent b2a2207 commit 48d93af

22 files changed

Lines changed: 1155 additions & 576 deletions

‎include/cppkafka/consumer.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -116,7 +116,7 @@ class CPPKAFKA_API Consumer : public KafkaHandleBase {
116116
Consumer& operator=(Consumer&&) = delete;
117117

118118
/**
119-
* \brief Closes and estroys the rdkafka handle
119+
* \brief Closes and destroys the rdkafka handle
120120
*
121121
* This will call Consumer::close before destroying the handle
122122
*/

‎include/cppkafka/cppkafka.h‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,8 @@
5656
#include <cppkafka/utils/buffered_producer.h>
5757
#include <cppkafka/utils/compacted_topic_processor.h>
5858
#include <cppkafka/utils/consumer_dispatcher.h>
59-
#include <cppkafka/utils/roundrobin_poll_adapter.h>
59+
#include <cppkafka/utils/poll_interface.h>
60+
#include <cppkafka/utils/poll_strategy_base.h>
61+
#include <cppkafka/utils/roundrobin_poll_strategy.h>
6062

6163
#endif

‎include/cppkafka/kafka_handle_base.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ class CPPKAFKA_API KafkaHandleBase {
7979
/**
8080
* \brief Resumes consumption/production from the given topic/partition list
8181
*
82-
* This translates into a call to rd_kafka_resume_partitions
82+
* This translates into a call to rd_kafka_resume_partitions
8383
*
8484
* \param topic_partitions The topic/partition list to resume consuming/producing from/to
8585
*/

‎include/cppkafka/utils/consumer_dispatcher.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -238,7 +238,7 @@ class CPPKAFKA_API BasicConsumerDispatcher {
238238
}
239239

240240
// Finds the first functor that accepts the parameters in a tuple and returns it. If no
241-
// such functor is found, a static asertion will occur
241+
// such functor is found, a static assertion will occur
242242
template <typename Tuple, typename... Functors>
243243
const typename find_type<Tuple, Functors...>::type&
244244
find_matching_functor(const Functors&... functors) {

‎include/cppkafka/utils/poll_interface.h‎

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Copyright (c) 2017, Matias Fontanini
3+
* All rights reserved.
4+
*
5+
* Redistribution and use in source and binary forms, with or without
6+
* modification, are permitted provided that the following conditions are
7+
* met:
8+
*
9+
* * Redistributions of source code must retain the above copyright
10+
* notice, this list of conditions and the following disclaimer.
11+
* * Redistributions in binary form must reproduce the above
12+
* copyright notice, this list of conditions and the following disclaimer
13+
* in the documentation and/or other materials provided with the
14+
* distribution.
15+
*
16+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27+
*
28+
*/
29+
30+
#ifndef CPPKAFKA_POLL_INTERFACE_H
31+
#define CPPKAFKA_POLL_INTERFACE_H
32+
33+
#include "../consumer.h"
34+
35+
namespace cppkafka {
36+
37+
/**
38+
* \interface PollInterface
39+
*
40+
* \brief Interface defining polling methods for the Consumer class
41+
*/
42+
struct PollInterface {
43+
virtual ~PollInterface() = default;
44+
45+
/**
46+
* \brief Get the underlying consumer controlled by this strategy
47+
*
48+
* \return A reference to the consumer instance
49+
*/
50+
virtual Consumer& get_consumer() = 0;
51+
52+
/**
53+
* \brief Sets the timeout for polling functions
54+
*
55+
* This calls Consumer::set_timeout
56+
*
57+
* \param timeout The timeout to be set
58+
*/
59+
virtual void set_timeout(std::chrono::milliseconds timeout) = 0;
60+
61+
/**
62+
* \brief Gets the timeout for polling functions
63+
*
64+
* This calls Consumer::get_timeout
65+
*
66+
* \return The timeout
67+
*/
68+
virtual std::chrono::milliseconds get_timeout() = 0;
69+
70+
/**
71+
* \brief Polls all assigned partitions for new messages in round-robin fashion
72+
*
73+
* Each call to poll() will first consume from the global event queue and if there are
74+
* no pending events, will attempt to consume from all partitions until a valid message is found.
75+
* The timeout used on this call will be the one configured via RoundRobinPollStrategy::set_timeout.
76+
*
77+
* \return A message. The returned message *might* be empty. It's necessary to check
78+
* that it's a valid one before using it (see example above).
79+
*
80+
* \remark You need to call poll() or poll_batch() periodically as a keep alive mechanism,
81+
* otherwise the broker will think this consumer is down and will trigger a rebalance
82+
* (if using dynamic subscription)
83+
*/
84+
virtual Message poll() = 0;
85+
86+
/**
87+
* \brief Polls for new messages
88+
*
89+
* Same as the other overload of RoundRobinPollStrategy::poll but the provided
90+
* timeout will be used instead of the one configured on this Consumer.
91+
*
92+
* \param timeout The timeout to be used on this call
93+
*/
94+
virtual Message poll(std::chrono::milliseconds timeout) = 0;
95+
96+
/**
97+
* \brief Polls all assigned partitions for a batch of new messages in round-robin fashion
98+
*
99+
* Each call to poll_batch() will first attempt to consume from the global event queue
100+
* and if the maximum batch number has not yet been filled, will attempt to fill it by
101+
* reading the remaining messages from each partition.
102+
*
103+
* \param max_batch_size The maximum amount of messages expected
104+
*
105+
* \return A list of messages
106+
*
107+
* \remark You need to call poll() or poll_batch() periodically as a keep alive mechanism,
108+
* otherwise the broker will think this consumer is down and will trigger a rebalance
109+
* (if using dynamic subscription)
110+
*/
111+
virtual MessageList poll_batch(size_t max_batch_size) = 0;
112+
113+
/**
114+
* \brief Polls all assigned partitions for a batch of new messages in round-robin fashion
115+
*
116+
* Same as the other overload of RoundRobinPollStrategy::poll_batch but the provided
117+
* timeout will be used instead of the one configured on this Consumer.
118+
*
119+
* \param max_batch_size The maximum amount of messages expected
120+
*
121+
* \param timeout The timeout for this operation
122+
*
123+
* \return A list of messages
124+
*/
125+
virtual MessageList poll_batch(size_t max_batch_size, std::chrono::milliseconds timeout) = 0;
126+
};
127+
128+
} //cppkafka
129+
130+
#endif //CPPKAFKA_POLL_INTERFACE_H

‎include/cppkafka/utils/poll_strategy_base.h‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
/*
2+
* Copyright (c) 2017, Matias Fontanini
3+
* All rights reserved.
4+
*
5+
* Redistribution and use in source and binary forms, with or without
6+
* modification, are permitted provided that the following conditions are
7+
* met:
8+
*
9+
* * Redistributions of source code must retain the above copyright
10+
* notice, this list of conditions and the following disclaimer.
11+
* * Redistributions in binary form must reproduce the above
12+
* copyright notice, this list of conditions and the following disclaimer
13+
* in the documentation and/or other materials provided with the
14+
* distribution.
15+
*
16+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
17+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
18+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
19+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
20+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
23+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
24+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
25+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
26+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
27+
*
28+
*/
29+
30+
#ifndef CPPKAFKA_POLL_STRATEGY_BASE_H
31+
#define CPPKAFKA_POLL_STRATEGY_BASE_H
32+
33+
#include <map>
34+
#include <boost/any.hpp>
35+
#include "../queue.h"
36+
#include "../topic_partition_list.h"
37+
#include "poll_interface.h"
38+
39+
namespace cppkafka {
40+
41+
/**
42+
* \brief Contains a partition queue and generic metadata which can be used to store
43+
* related (user-specific) information.
44+
*/
45+
struct QueueData {
46+
Queue queue_;
47+
boost::any metadata_;
48+
};
49+
50+
/**
51+
* \class PollStrategyBase
52+
*
53+
* \brief Base implementation of the PollInterface
54+
*/
55+
class PollStrategyBase : public PollInterface
56+
{
57+
public:
58+
using QueueMap = std::map<TopicPartition, QueueData>;
59+
60+
/**
61+
* \brief Constructor
62+
*
63+
* \param consumer A reference to the polled consumer instance
64+
*/
65+
explicit PollStrategyBase(Consumer& consumer);
66+
67+
/**
68+
* \brief Destructor
69+
*/
70+
~PollStrategyBase();
71+
72+
/**
73+
* \sa PollInterface::set_timeout
74+
*/
75+
void set_timeout(std::chrono::milliseconds timeout) override;
76+
77+
/**
78+
* \sa PollInterface::get_timeout
79+
*/
80+
std::chrono::milliseconds get_timeout() override;
81+
82+
/**
83+
* \sa PollInterface::get_consumer
84+
*/
85+
Consumer& get_consumer() final;
86+
87+
protected:
88+
/**
89+
* \brief Get the queues from all assigned partitions
90+
*
91+
* \return A map of queues indexed by partition
92+
*/
93+
QueueMap& get_partition_queues();
94+
95+
/**
96+
* \brief Get the main consumer queue which services the underlying Consumer object
97+
*
98+
* \return The consumer queue
99+
*/
100+
QueueData& get_consumer_queue();
101+
102+
/**
103+
* \brief Return the next queue to be processed
104+
*
105+
* Depending on the polling strategy, each implementation must define it's own algorithm for
106+
* determining the next queue to poll.
107+
*
108+
* \return A partition queue
109+
*/
110+
virtual QueueData& get_next_queue() = 0;
111+
112+
/**
113+
* \brief Reset the internal state of the queues.
114+
*
115+
* Use this function to reset the state of any polling strategy or algorithm.
116+
*
117+
* \remark This function gets called by on_assignement(), on_revocation() and on_rebalance_error()
118+
*/
119+
virtual void reset_state();
120+
121+
/**
122+
* \brief Function to be called when a new partition assignment takes place
123+
*
124+
* This method contains a default implementation. It adds all the new queues belonging
125+
* to the provided partition list and calls reset_state().
126+
*
127+
* \param partitions Assigned topic partitions
128+
*/
129+
virtual void on_assignment(TopicPartitionList& partitions);
130+
131+
/**
132+
* \brief Function to be called when an old partition assignment gets revoked
133+
*
134+
* This method contains a default implementation. It removes all the queues
135+
* belonging to the provided partition list and calls reset_state().
136+
*
137+
* \param partitions Revoked topic partitions
138+
*/
139+
virtual void on_revocation(const TopicPartitionList& partitions);
140+
141+
/**
142+
* \brief Function to be called when a topic rebalance error happens
143+
*
144+
* This method contains a default implementation. Calls reset_state().
145+
*
146+
* \param error The rebalance error
147+
*/
148+
virtual void on_rebalance_error(Error error);
149+
150+
private:
151+
Consumer& consumer_;
152+
QueueData consumer_queue_;
153+
QueueMap partition_queues_;
154+
Consumer::AssignmentCallback assignment_callback_;
155+
Consumer::RevocationCallback revocation_callback_;
156+
Consumer::RebalanceErrorCallback rebalance_error_callback_;
157+
};
158+
159+
} //cppkafka
160+
161+
#endif //CPPKAFKA_POLL_STRATEGY_BASE_H

0 commit comments

Comments
 (0)