Made num_partitions optional (#1515)

* Default New Topic Params -> num_partitions optional

* Added to Changelog

* Keeping one source of truth

* Made unit tests and Integration bounded to scope of python client

* Changelog

* Unit tests removed from Integration test

* Flake8

* Review Comments Addressed

* flake8 and whitespace changes

* Whitespace Changes

* flake8

* Whitespaces

* Whitespace

* PR comment addressal

* PR comment addressal

* PR comment addressal

* PR comment Addressal

Author: mahajanadhitya

CHANGELOG.md

@@ -5,7 +5,7 @@
 - Added `set_sasl_credentials`. This new method (on the Producer, Consumer, and AdminClient) allows modifying the stored
   SASL PLAIN/SCRAM credentials that will be used for subsequent (new) connections to a broker (#1511).
 - Wheels for Linux / arm64 (#1496).
-
+- Added support for Default num_partitions in CreateTopics Admin API.
 
 ## v2.0.2
 

src/confluent_kafka/src/Admin.c

@@ -418,6 +418,7 @@ static PyObject *Admin_create_topics (Handle *self, PyObject *args,
         rd_kafka_AdminOptions_t *c_options = NULL;
         int tcnt;
         int i;
+        int topic_partition_count;
         rd_kafka_NewTopic_t **c_objs;
         rd_kafka_queue_t *rkqu;
         CallState cs;
@@ -492,10 +493,16 @@ static PyObject *Admin_create_topics (Handle *self, PyObject *args,
                                 goto err;
                         }
 
+                        if (newt->num_partitions == -1) {
+                                topic_partition_count = PyList_Size(newt->replica_assignment);
+                        } else {
+                                topic_partition_count = newt->num_partitions;
+                        }
                         if (!Admin_set_replica_assignment(
                                     "CreateTopics", (void *)c_objs[i],
                                     newt->replica_assignment,
-                                    newt->num_partitions, newt->num_partitions,
+                                    topic_partition_count,
+                                    topic_partition_count, 
                                     "num_partitions")) {
                                 i++;
                                 goto err;

src/confluent_kafka/src/AdminTypes.c

@@ -74,18 +74,20 @@ static int NewTopic_init (PyObject *self0, PyObject *args,
                                "config",
                                NULL };
 
+        self->num_partitions = -1;
         self->replication_factor = -1;
         self->replica_assignment = NULL;
         self->config = NULL;
 
-        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "si|iOO", kws,
+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|iiOO", kws,
                                          &topic, &self->num_partitions,
                                          &self->replication_factor,
                                          &self->replica_assignment,
                                          &self->config))
                 return -1;
 
 
+
         if (self->config) {
                 if (!PyDict_Check(self->config)) {
                         PyErr_SetString(PyExc_TypeError,
@@ -125,7 +127,8 @@ static PyMemberDef NewTopic_members[] = {
         { "topic", T_STRING, offsetof(NewTopic, topic), READONLY,
           ":py:attribute:topic - Topic name (string)" },
         { "num_partitions", T_INT, offsetof(NewTopic, num_partitions), 0,
-          ":py:attribute: Number of partitions (int)" },
+          ":py:attribute: Number of partitions (int).\n"
+          "Or -1 if a replica_assignment is specified" },
         { "replication_factor", T_INT, offsetof(NewTopic, replication_factor),
           0,
           " :py:attribute: Replication factor (int).\n"
@@ -147,6 +150,11 @@ static PyMemberDef NewTopic_members[] = {
 
 
 static PyObject *NewTopic_str0 (NewTopic *self) {
+        if (self->num_partitions == -1) {
+                return cfl_PyUnistr(
+                _FromFormat("NewTopic(topic=%s)",
+                            self->topic));
+        }
         return cfl_PyUnistr(
                 _FromFormat("NewTopic(topic=%s,num_partitions=%d)",
                             self->topic, self->num_partitions));
@@ -202,7 +210,12 @@ NewTopic_richcompare (NewTopic *self, PyObject *o2, int op) {
 
 static long NewTopic_hash (NewTopic *self) {
         PyObject *topic = cfl_PyUnistr(_FromString(self->topic));
-        long r = PyObject_Hash(topic) ^ self->num_partitions;
+        long r;
+        if (self->num_partitions == -1) {
+                r = PyObject_Hash(topic);
+        } else {
+                r = PyObject_Hash(topic) ^ self->num_partitions;
+        }
         Py_DECREF(topic);
         return r;
 }
@@ -233,12 +246,12 @@ PyTypeObject NewTopicType = {
         "NewTopic specifies per-topic settings for passing to "
         "AdminClient.create_topics().\n"
         "\n"
-        ".. py:function:: NewTopic(topic, num_partitions, [replication_factor], [replica_assignment], [config])\n"
+        ".. py:function:: NewTopic(topic, [num_partitions], [replication_factor], [replica_assignment], [config])\n"
         "\n"
         "  Instantiate a NewTopic object.\n"
         "\n"
         "  :param string topic: Topic name\n"
-        "  :param int num_partitions: Number of partitions to create\n"
+        "  :param int num_partitions: Number of partitions to create, or -1 if replica_assignment is used.\n"
         "  :param int replication_factor: Replication factor of partitions, or -1 if replica_assignment is used.\n"
         "  :param list replica_assignment: List of lists with the replication assignment for each new partition.\n"
         "  :param dict config: Dict (str:str) of topic configuration. See http://kafka.apache.org/documentation.html#topicconfigs\n"

tests/test_Admin.py

@@ -122,6 +122,11 @@ def test_create_topics_api():
     with pytest.raises(Exception):
         a.create_topics([None, NewTopic("mytopic", 1, 2)])
 
+    try:
+        a.create_topics([NewTopic("mytopic")])
+    except Exception as err:
+        assert False, f"When none of the partitions, \
+            replication and assignment is present, the request should not fail, but it does with error {err}"
     fs = a.create_topics([NewTopic("mytopic", 3, 2)])
     with pytest.raises(KafkaException):
         for f in concurrent.futures.as_completed(iter(fs.values())):