Skip to content

Commit 13bddf3

Browse files
kenneth-jiakenneth-jia
committed
Add validation for Properties
1 parent 30ec31a commit 13bddf3

File tree

3 files changed

+119
-16
lines changed

3 files changed

+119
-16
lines changed

‎README.md

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -187,7 +187,7 @@ Eventually, we worked out the ***modern-cpp-kafka***, -- a ***header-only*** lib
187187
* Kafka cluster setup
188188

189189
* [Quick Start For Cluster Setup](https://kafka.apache.org/documentation/#quickstart)
190-
190+
191191
* [Cluster Setup Scripts For Test](https://github.com/morganstanley/modern-cpp-kafka/blob/main/scripts/start-local-kafka-cluster.py)
192192

193193
* [Kafka Broker Configuration](doc/KafkaBrokerConfiguration.md)
@@ -200,12 +200,13 @@ Eventually, we worked out the ***modern-cpp-kafka***, -- a ***header-only*** lib
200200
| `KAFKA_BROKER_PIDS` | The broker PIDs for test runner to manipulate | `export KAFKA_BROKER_PIDS=61567,61569,61571` |
201201
| `KAFKA_CLIENT_ADDITIONAL_SETTINGS` | Could be used for addtional configuration for Kafka clients | `export KAFKA_CLIENT_ADDITIONAL_SETTINGS="security.protocol=SASL_PLAINTEXT;sasl.kerberos.service.name=...;sasl.kerberos.keytab=...;sasl.kerberos.principal=..."` |
202202

203-
* The environment variable `KAFKA_BROKER_LIST` is mandatory for integration/robustness test
203+
* The environment variable `KAFKA_BROKER_LIST` is mandatory for integration/robustness test, which requires the Kafka cluster.
204+
205+
* The environment variable `KAFKA_BROKER_PIDS` is mandatory for robustness test, which requires the Kafka cluster and the privilege to stop/resume the brokers.
204206

205-
* The environment variable `KAFKA_BROKER_PIDS` is mandatory for robustness test
207+
| Test Type | `KAFKA_BROKER_LIST` | `KAFKA_BROKER_PIDS` |
208+
| -------------------------------------------------------------------------------------------------- | -------------------- | ------------------- |
209+
| [tests/unit](https://github.com/morganstanley/modern-cpp-kafka/tree/main/tests/unit) | - | - |
210+
| [tests/integration](https://github.com/morganstanley/modern-cpp-kafka/tree/main/tests/integration) | Required | - |
211+
| [tests/robustness](https://github.com/morganstanley/modern-cpp-kafka/tree/main/tests/robustness) | Required | Required |
206212

207-
| Test Type | Requires Kafka Cluster | Requires Privilege to Stop/Resume the Brokers |
208-
| -------------------------------------------------------------------------------------------------- | ------------------------ | --------------------------------------------- |
209-
| [tests/unit](https://github.com/morganstanley/modern-cpp-kafka/tree/main/tests/unit) | - | - |
210-
| [tests/integration](https://github.com/morganstanley/modern-cpp-kafka/tree/main/tests/integration) | Y (`KAFKA_BROKER_LIST`) | - |
211-
| [tests/robustness`](https://github.com/morganstanley/modern-cpp-kafka/tree/main/tests/robustness) | Y (`KAFKA_BROKER_LIST`) | Y (`KAFKA_BROKER_PIDS`) |

‎include/kafka/Properties.h

Lines changed: 46 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,21 @@ class Properties
4242
template<class T>
4343
static std::string getString(const std::string& value) { return value; }
4444

45+
const ValueType& validate(const std::string& key) const
46+
{
47+
static const std::vector<std::string> nonStringValueKeys = {
48+
"log_cb", "error_cb", "stats_cb", "oauthbearer_token_refresh_cb", "interceptors"
49+
};
50+
51+
if ((expectedKey.empty() && std::any_of(nonStringValueKeys.cbegin(), nonStringValueKeys.cend(), [key](const auto& k) { return k == key; }))
52+
|| (!expectedKey.empty() && key != expectedKey))
53+
{
54+
throw std::runtime_error("Invalid key/value for configuration: " + key);
55+
}
56+
57+
return *this;
58+
}
59+
4560
template<class T>
4661
struct ObjWrap: public Object
4762
{
@@ -55,18 +70,35 @@ class Properties
5570

5671
ValueType() = default;
5772

58-
ValueType(const std::string& value) { object = std::make_shared<ObjWrap<std::string>>(value); } // NOLINT
59-
ValueType(const LogCallback& cb) { object = std::make_shared<ObjWrap<LogCallback>>(cb); } // NOLINT
60-
ValueType(const ErrorCallback& cb) { object = std::make_shared<ObjWrap<ErrorCallback>>(cb); } // NOLINT
61-
ValueType(const StatsCallback& cb) { object = std::make_shared<ObjWrap<StatsCallback>>(cb); } // NOLINT
62-
ValueType(const OauthbearerTokenRefreshCallback& cb) { object = std::make_shared<ObjWrap<OauthbearerTokenRefreshCallback>>(cb); } // NOLINT
63-
ValueType(const Interceptors& interceptors) { object = std::make_shared<ObjWrap<Interceptors>>(interceptors); } // NOLINT
73+
ValueType(const std::string& value) // NOLINT
74+
{ object = std::make_shared<ObjWrap<std::string>>(value); }
75+
76+
ValueType(const LogCallback& cb) // NOLINT
77+
: expectedKey("log_cb")
78+
{ object = std::make_shared<ObjWrap<LogCallback>>(cb); }
79+
80+
ValueType(const ErrorCallback& cb) // NOLINT
81+
: expectedKey("error_cb")
82+
{ object = std::make_shared<ObjWrap<ErrorCallback>>(cb); }
83+
84+
ValueType(const StatsCallback& cb) // NOLINT
85+
: expectedKey("stats_cb")
86+
{ object = std::make_shared<ObjWrap<StatsCallback>>(cb); }
87+
88+
ValueType(const OauthbearerTokenRefreshCallback& cb) // NOLINT
89+
: expectedKey("oauthbearer_token_refresh_cb")
90+
{ object = std::make_shared<ObjWrap<OauthbearerTokenRefreshCallback>>(cb); }
91+
92+
ValueType(const Interceptors& interceptors) // NOLINT
93+
: expectedKey("interceptors")
94+
{ object = std::make_shared<ObjWrap<Interceptors>>(interceptors); }
6495

6596
bool operator==(const ValueType& rhs) const { return toString() == rhs.toString(); }
6697

6798
std::string toString() const { return object->toString(); }
6899

69100
private:
101+
std::string expectedKey;
70102
std::shared_ptr<Object> object;
71103
};
72104

@@ -76,7 +108,13 @@ class Properties
76108

77109
Properties() = default;
78110
Properties(const Properties&) = default;
79-
Properties(PropertiesMap kvMap): _kvMap(std::move(kvMap)) {} // NOLINT
111+
Properties(PropertiesMap kvMap): _kvMap(std::move(kvMap)) // NOLINT
112+
{
113+
for (const auto& kv: _kvMap)
114+
{
115+
kv.second.validate(kv.first);
116+
}
117+
}
80118
virtual ~Properties() = default;
81119

82120
bool operator==(const Properties& rhs) const { return map() == rhs.map(); }
@@ -88,7 +126,7 @@ class Properties
88126
template <class T>
89127
Properties& put(const std::string& key, const T& value)
90128
{
91-
_kvMap[key] = ValueType(value);
129+
_kvMap[key] = ValueType(value).validate(key);
92130
return *this;
93131
}
94132

‎tests/unit/TestProperties.cc

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -126,3 +126,67 @@ TEST(Properties, SensitiveProperties)
126126

127127
EXPECT_EQ("sasl.password=*|sasl.username=*|ssl.key.password=*|ssl.key.pem=*|ssl.keystore.password=*|ssl_key=*", props.toString());
128128
}
129+
130+
TEST(Properties, Validation)
131+
{
132+
kafka::Properties props;
133+
134+
props.put("whatever", "somevalue");
135+
136+
// Test with invalid keys
137+
auto tryWithInvalidKey = [&props](auto v)
138+
{
139+
try
140+
{
141+
props.put("invalid_key", v);
142+
return false;
143+
}
144+
catch (const std::runtime_error& e)
145+
{
146+
std::cout << "Exception caught: " << e.what() << std::endl;
147+
}
148+
return true;
149+
};
150+
151+
EXPECT_TRUE(tryWithInvalidKey([](int /*level*/, const char* /*filename*/, int /*lineno*/, const char* msg) { std::cout << msg << std::endl; }));
152+
EXPECT_TRUE(tryWithInvalidKey([](const kafka::Error& err) { std::cerr << err.toString() << std::endl; }));
153+
EXPECT_TRUE(tryWithInvalidKey([](const std::string& stats) { std::cout << stats << std::endl; }));
154+
const kafka::clients::OauthbearerTokenRefreshCallback oauthTokenRefreshCb = [](const std::string&) { return kafka::clients::SaslOauthbearerToken(); };
155+
EXPECT_TRUE(tryWithInvalidKey(oauthTokenRefreshCb));
156+
EXPECT_TRUE(tryWithInvalidKey(kafka::clients::Interceptors{}));
157+
158+
// Test with invalid values
159+
const auto tryWithInvalidValue = [&props](const std::string& key)
160+
{
161+
try
162+
{
163+
props.put(key, "haha");
164+
return false;
165+
}
166+
catch (const std::runtime_error& e)
167+
{
168+
std::cout << "exception caught: " << e.what() << std::endl;
169+
}
170+
return true;
171+
};
172+
173+
EXPECT_TRUE(tryWithInvalidValue(kafka::clients::Config::LOG_CB));
174+
EXPECT_TRUE(tryWithInvalidValue(kafka::clients::Config::ERROR_CB));
175+
EXPECT_TRUE(tryWithInvalidValue(kafka::clients::Config::STATS_CB));
176+
EXPECT_TRUE(tryWithInvalidValue(kafka::clients::Config::OAUTHBEARER_TOKEN_REFRESH_CB));
177+
EXPECT_TRUE(tryWithInvalidValue(kafka::clients::Config::INTERCEPTORS));
178+
179+
// Failure within constructor
180+
try
181+
{
182+
const kafka::Properties properties = {{
183+
{ "interceptorsxx", { kafka::clients::Interceptors{} } },
184+
}};
185+
EXPECT_FALSE(true);
186+
}
187+
catch (const std::runtime_error& e)
188+
{
189+
std::cout << "exception caught: " << e.what() << std::endl;
190+
}
191+
}
192+

0 commit comments

Comments
 (0)