Run tests through kafka #2005

Ark2307 · 2025-01-21T07:55:31Z

No description provided.

ayushaga14 · 2025-01-21T11:32:18Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingConsumer.java

+        ObjectId testingRunResultSummaryId = new ObjectId(jsonObject.getString("testingRunResultSummaryId"));
+        ApiInfo.ApiInfoKey apiInfoKey = ApiInfo.getApiInfoKeyFromString(jsonObject.getString("apiInfoKey"));
+        String subcategory = jsonObject.getString("subcategory");
+        List<TestingRunResult.TestLog> testLogs = JSON.parseArray(jsonObject.getString("testLogs"), TestingRunResult.TestLog.class);


check how testlogs are used and stored

ayushaga14 · 2025-01-21T11:32:55Z

libs/dao/src/main/java/com/akto/dto/testing/info/TestMessages.java

+import com.akto.dto.ApiInfo.ApiInfoKey;
+import com.akto.dto.testing.TestingRunResult;
+
+public class TestMessages {


SingleTestPayload

ayushaga14 · 2025-01-21T11:34:25Z

libs/dao/src/main/java/com/akto/util/Constants.java

@@ -23,6 +23,13 @@ private Constants() {}
    public static final String AKTO_NODE_ID = "x-akto-node";
    public static final String AKTO_REMOVE_AUTH= "x-akto-remove-auth";

+    public static final String LOCAL_KAFKA_BROKER_URL = "localhost:29092"; // run kafka process with name kafka1 in docker


don't hardcode kafka related variables

ayushaga14 · 2025-01-21T11:34:53Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingProducer.java

@@ -0,0 +1,99 @@
+package com.akto.testing.testing_with_kafka;


package name is really weird, rename

rename classes to producer.java, consumer.java

ayushaga14 · 2025-01-21T11:35:35Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/CommonSingletonForTesting.java

+import com.akto.dto.testing.TestingRunConfig;
+import com.akto.store.TestingUtil;
+
+public class CommonSingletonForTesting {


rename this

ayushaga14 · 2025-01-21T11:36:49Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingProducer.java

+        }
+    }
+
+    private boolean isKafkaEmpty(){


is this intentional?

ayushaga14 · 2025-01-21T11:44:58Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingProducer.java

+        return null;
+    }
+
+    private static void deleteAllMessagesFromTopic(String bootstrapServers, String topicName)


delete the topic itself, new message push wll create a new one

ayushaga14 · 2025-01-21T11:46:19Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingProducer.java

+
+    public static final Kafka producer = new Kafka(Constants.LOCAL_KAFKA_BROKER_URL, 500, 1000);
+
+    public static Void pushMessagesToKafka(List<TestMessages> messages){


ayushaga14 · 2025-01-27T10:24:07Z

apps/testing/src/main/java/com/akto/testing/Main.java

+
+            String testingRunId = currentTestInfo.getString("testingRunId");
+            String testingRunSummaryId = currentTestInfo.getString("summaryId");
+            TestingRun testingRun = TestingRunDao.instance.findOne(Filters.eq(Constants.ID, new ObjectId(testingRunId)));


review this and add try catch, if something panics, whole process will exit

apps/testing/src/main/java/com/akto/testing/TestExecutor.java

ayushaga14 · 2025-01-27T11:09:23Z

apps/testing/src/main/java/com/akto/testing/TestExecutor.java

+        return null;
+    }
+
+    private Void doJobForTest(int accountId, String testSubCategory, ApiInfo.ApiInfoKey apiInfoKey,


which job? rename

ayushaga14 · 2025-01-27T11:17:54Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingConsumer.java

+                    executor.awaitTermination(maxRunTimeInSeconds, TimeUnit.SECONDS);
+                    break;
+                }
+                Thread.sleep(100);


100 seconds?

ayushaga14 · 2025-01-27T11:19:00Z

apps/testing/src/main/java/com/akto/testing/testing_with_kafka/TestingConsumer.java

+                    try {
+                        future.get(4, TimeUnit.MINUTES); 
+                    } catch (InterruptedException e) {
+                        logger.error("Task timed out: " + message);


is there a way to know which message timed out?

ayushaga14

resolve comments

Ark2307 added 3 commits January 16, 2025 20:07

Moved around code to create basic structure

cc34d40

Completing basic code to run tests from consumer

0225556

Adding file check and completing happy case of testing

f6a8ff4

Ark2307 force-pushed the feature/run_tests_through_kafka branch from 3a99143 to bcf7528 Compare January 21, 2025 11:22

ayushaga14 reviewed Jan 21, 2025

View reviewed changes

Ark2307 force-pushed the feature/run_tests_through_kafka branch from bcf7528 to f6a8ff4 Compare January 21, 2025 11:45

ayushaga14 reviewed Jan 21, 2025

View reviewed changes

Ark2307 added 9 commits January 21, 2025 20:35

Code clean up

c2bd77b

Merge branch 'master' into feature/run_tests_through_kafka

f778717

ToDo: stop test

1d38495

Fixed using parallel consumer

005b9e8

Final commit to mark all the cases done

6d520cd

Temp commit for staging

5b4f425

Added slack alert and fixed threadpool init

61d5116

Merge branch 'master' into feature/run_tests_through_kafka

40c58d3

Added support for backward compatibility

896d69f

ayushaga14 reviewed Jan 27, 2025

View reviewed changes

apps/testing/src/main/java/com/akto/testing/TestExecutor.java Show resolved Hide resolved

ayushaga14 reviewed Jan 27, 2025

View reviewed changes

apps/testing/src/main/java/com/akto/testing/TestExecutor.java Show resolved Hide resolved

ayushaga14 reviewed Jan 27, 2025

View reviewed changes

ayushaga14 requested changes Jan 27, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Run tests through kafka #2005

Run tests through kafka #2005

Ark2307 commented Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 21, 2025

ayushaga14 Jan 27, 2025

ayushaga14 Jan 27, 2025

ayushaga14 Jan 27, 2025

Ark2307 Jan 27, 2025

ayushaga14 Jan 27, 2025

ayushaga14 left a comment

		@@ -0,0 +1,99 @@
		package com.akto.testing.testing_with_kafka;


		public static final Kafka producer = new Kafka(Constants.LOCAL_KAFKA_BROKER_URL, 500, 1000);

		public static Void pushMessagesToKafka(List<TestMessages> messages){

Run tests through kafka #2005

Are you sure you want to change the base?

Run tests through kafka #2005

Conversation

Ark2307 commented Jan 21, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ayushaga14 left a comment

Choose a reason for hiding this comment