Add protobuf test cases, add schema registry to workflows.

Author: Vihas Splunk

.github/workflows/ci_build_test.yaml

@@ -149,7 +149,6 @@ jobs:
       CI_KAFKA_HEADER_INDEX: kafka
       CI_DATAGEN_IMAGE: rock1017/log-generator:latest
       CI_OLD_CONNECTOR_VERSION: v2.0.1
-      SCHEMA_REGISTRY_URL: ${{ Secrets.SCHEMA_REGISTRY_URL }}
 
     steps:
       - name: Checkout
@@ -222,6 +221,18 @@ jobs:
           name: splunk-kafka-connector
           path: /tmp
 
+      - name: Up the Schema Registry
+        run: |
+          cd /tmp && wget https://packages.confluent.io/archive/7.1/confluent-community-7.1.1.tar.gz
+          sudo tar xzf confluent-community-7.1.1.tar.gz
+          cd confluent-7.1.1
+          bin/schema-registry-start ./etc/schema-registry/schema-registry.properties &
+
+      - name: Register the protobuf schema
+        run: |
+          sleep 10
+          curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" --data "{\"schemaType\": \"PROTOBUF\",\"schema\": \"syntax = \\\"proto3\\\";\\npackage com.mycorp.mynamespace;\\n\\nmessage MyRecord {\\n  string id = 1;\\n  float amount = 2;\\n  string customer_id = 3;\\n}\\n\"}" http://localhost:8081/subjects/prototopic-value/versions
+
       - name: Test kafka connect upgrade
         run: |
           echo "Download kafka connect "$CI_OLD_CONNECTOR_VERSION

src/main/java/com/splunk/kafka/connect/SplunkSinkConnectorConfig.java

@@ -318,7 +318,7 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
         timestampFormat = getString(TIMESTAMP_FORMAT_CONF).trim();
         validateRegexForTimestamp(regex);
         queueCapacity = getInt(QUEUE_CAPACITY_CONF);
-      
+        validateQueueCapacity(queueCapacity);
     }
 
 
@@ -558,6 +558,12 @@ private void validateRegexForTimestamp(String regex) {
         }
     }
 
+    private void validateQueueCapacity(int queueCapacity) {
+        if (queueCapacity <= 0) {
+            throw new ConfigException("queue capacity should be greater than " + queueCapacity);
+        }
+    }
+
     private static boolean getNamedGroupCandidates(String regex) {
         Matcher m = Pattern.compile("\\(\\?<([a-zA-Z][a-zA-Z0-9]*)>").matcher(regex);
         while (m.find()) {

src/test/java/com/splunk/kafka/connect/SplunkSinkConnecterTest.java

@@ -261,6 +261,34 @@ public void testInvalidSplunkConfigurationsWithValidationEnabled() {
         Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
     }
 
+    @Test
+    public void testValidQueueCapacity() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SplunkSinkConnector connector = new SplunkSinkConnector();
+        configs.put("splunk.hec.concurrent.queue.capacity", "100");
+        configs.put("topics", "b");
+        configs.put("splunk.indexes", "b");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        clientInstance.client.setResponse(CloseableHttpClientMock.success);
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertDoesNotThrow(()->connector.validate(configs));
+    }
+
+    @Test
+    public void testInvalidQueueCapacity() {
+        final Map<String, String> configs = new HashMap<>();
+        addNecessaryConfigs(configs);
+        SplunkSinkConnector connector = new SplunkSinkConnector();
+        configs.put("splunk.hec.concurrent.queue.capacity", "-1");
+        configs.put("topics", "b");
+        configs.put("splunk.indexes", "b");
+        MockHecClientWrapper clientInstance = new MockHecClientWrapper();
+        clientInstance.client.setResponse(CloseableHttpClientMock.success);
+        ((SplunkSinkConnector) connector).setHecInstance(clientInstance);
+        Assertions.assertThrows(ConfigException.class, ()->connector.validate(configs));
+    }
+
     private void addNecessaryConfigs(Map<String, String> configs) {
         configs.put(URI_CONF, TEST_URI);
         configs.put(TOKEN_CONF, "blah");

test/conftest.py

@@ -70,7 +70,7 @@ def pytest_configure():
     producer.send("test_splunk_hec_malformed_events", {})
     producer.send("test_splunk_hec_malformed_events", {"&&": "null", "message": ["$$$$****////", 123, None]})
     producer.send("record_key",{"timestamp": config['timestamp']},b"{}")
-    protobuf_producer.send("prototopic",value=b'\x00\x00\x00\x00\x01\x00\n\x011\x12\r10-01-04-3:45\x18\x15%\x00\x00*C*\x02No:\x12\n\x011\x12\x04this\x1a\x07New oneB\x0c\n\x011\x12\x07shampooJ\x04Many')
+    protobuf_producer.send("prototopic",value=b'\x00\x00\x00\x00\x01\x00\n\x011\x1533\xf3?\x1a\x05Hello')
     timestamp_producer.send("date_format",b"{\"id\": \"19\",\"host\":\"host-01\",\"source\":\"bu\",\"fields\":{\"hn\":\"hostname\",\"CLASS\":\"class\",\"cust_id\":\"000013934\",\"time\": \"Jun 13 2010 23:11:52.454 UTC\",\"category\":\"IFdata\",\"ifname\":\"LoopBack7\",\"IFdata.Bits received\":\"0\",\"IFdata.Bits sent\":\"0\"}")
     timestamp_producer.send("epoch_format",b"{\"id\": \"19\",\"host\":\"host-01\",\"source\":\"bu\",\"fields\":{\"hn\":\"hostname\",\"CLASS\":\"class\",\"cust_id\":\"000013934\",\"time\": \"1555209605000\",\"category\":\"IFdata\",\"ifname\":\"LoopBack7\",\"IFdata.Bits received\":\"0\",\"IFdata.Bits sent\":\"0\"}")
     producer.flush()

test/lib/connect_params.py

@@ -188,7 +188,7 @@
      "splunk_hec_raw": False,
      "topics": "prototopic",
      "value_converter":"io.confluent.connect.protobuf.ProtobufConverter",
-     "value_converter_schema_registry_url": os.environ["SCHEMA_REGISTRY_URL"],
+     "value_converter_schema_registry_url": "http://localhost:8081",
      "value_converter_schemas_enable":"true"
     },
     {"name": "test_extracted_timestamp_dateformat",

test/requirements.txt

@@ -1,5 +1,5 @@
 pytest
-requests
+requests == 2.28.2
 kafka-python
 pyyaml == 5.4.1
 jinja2