You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by "umustafi (via GitHub)" <gi...@apache.org> on 2023/11/01 21:40:56 UTC
[PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
umustafi opened a new pull request, #3817:
URL: https://github.com/apache/gobblin/pull/3817
Dear Gobblin maintainers,
Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!
### JIRA
- [X] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1946
### Description
- [X] Here are some details about my PR, including screenshots (if applicable):
Add basic unit testing framework for `DagActionStoreChangeMonitor.processMessage()` function which is used for processing `launch`, `kill`, and `resume` flow action events. This is a crucial piece of functionality used to launch events, so it needs to be resilient to malformed message types and handle them without interrupting the thread processing the queue.
### Tests
- [X] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
Tests calling process message on HEARTBEAT, INSERT, UPDATE, DELETE operation types. The main functionality we seek to test is to ensure heartbeat type messages or messages with null dag action types are handled without throwing a `NullPointerException`. The unit tests are rudimentary other than the aforementioned functionality, mocking all related classes so their states and the actual result of processing the message is not validated at the moment.
### Commits
- [X] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1379405036
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
+ }
+
+ @Override
+ protected void startUp() {
+ super.startUp();
+ }
+ }
+
+ MockDagActionStoreChangeMonitor createMockDagActionStoreChangeMonitor() {
+ Config config = ConfigFactory.empty().withValue(ConfigurationKeys.KAFKA_BROKERS, ConfigValueFactory.fromAnyRef("localhost:0000"))
+ .withValue(Kafka09ConsumerClient.GOBBLIN_CONFIG_VALUE_DESERIALIZER_CLASS_KEY, ConfigValueFactory.fromAnyRef("org.apache.kafka.common.serialization.ByteArrayDeserializer"))
+ .withValue(ConfigurationKeys.STATE_STORE_ROOT_DIR_KEY, ConfigValueFactory.fromAnyRef("/tmp/fakeStateStore"))
+ .withValue("zookeeper.connect", ConfigValueFactory.fromAnyRef("localhost:2121"));
+ return new MockDagActionStoreChangeMonitor("dummyTopic", config, 5, true);
+ }
+
+ @BeforeClass
+ public void setup() {
+ mockDagActionStoreChangeMonitor = createMockDagActionStoreChangeMonitor();
+ mockDagActionStoreChangeMonitor.startUp();
+ }
+
+ /**
+ * Ensure no NPE results from passing a HEARTBEAT type message with a null {@link DagActionValue}
+ */
+ @Test
+ public void testProcessMessageWithHeartbeat() {
+ Kafka09ConsumerClient.Kafka09ConsumerRecord consumerRecord =
+ wrapDagActionStoreChangeEvent(OperationType.HEARTBEAT, "", "", "", null);
+ mockDagActionStoreChangeMonitor.processMessage(consumerRecord);
+ }
+
+ /**
+ * Tests process message with an INSERT type message
+ */
+ @Test
+ public void testProcessMessageWithInsert() {
+ Kafka09ConsumerClient.Kafka09ConsumerRecord consumerRecord =
+ wrapDagActionStoreChangeEvent(OperationType.INSERT, FLOW_GROUP, FLOW_NAME, FLOW_EXECUTION_ID, DagActionValue.LAUNCH);
+ mockDagActionStoreChangeMonitor.processMessage(consumerRecord);
Review Comment:
Do we just need to test that `processMessage` returns without any errors, or can we verify any other stuff too? I havent looked what all `processMessage` do
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380799356
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
Review Comment:
The methods are protected access but these are in a diff module so override it only to access the method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#issuecomment-1799999830
Tests passing on own fork https://github.com/umustafi/gobblin/pull/10/checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1385991904
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
Review Comment:
looks better now. maybe we do not need `super.` there
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1387015684
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
Review Comment:
I think this error would occur on the producer side, not on the consumer side
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1379391670
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
Review Comment:
Why to override when all the method do is call super's method? same for the `startUp`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380829614
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
Review Comment:
These are messages all containing only a few fields, so I don't expect this case but I can test it if I find a way to create. What would the too large message cause as an error? How can we test?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo merged PR #3817:
URL: https://github.com/apache/gobblin/pull/3817
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380830610
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
+ }
+
+ @Override
+ protected void startUp() {
+ super.startUp();
+ }
+ }
+
+ MockDagActionStoreChangeMonitor createMockDagActionStoreChangeMonitor() {
+ Config config = ConfigFactory.empty().withValue(ConfigurationKeys.KAFKA_BROKERS, ConfigValueFactory.fromAnyRef("localhost:0000"))
+ .withValue(Kafka09ConsumerClient.GOBBLIN_CONFIG_VALUE_DESERIALIZER_CLASS_KEY, ConfigValueFactory.fromAnyRef("org.apache.kafka.common.serialization.ByteArrayDeserializer"))
+ .withValue(ConfigurationKeys.STATE_STORE_ROOT_DIR_KEY, ConfigValueFactory.fromAnyRef("/tmp/fakeStateStore"))
+ .withValue("zookeeper.connect", ConfigValueFactory.fromAnyRef("localhost:2121"));
+ return new MockDagActionStoreChangeMonitor("dummyTopic", config, 5, true);
+ }
+
+ @BeforeClass
+ public void setup() {
+ mockDagActionStoreChangeMonitor = createMockDagActionStoreChangeMonitor();
+ mockDagActionStoreChangeMonitor.startUp();
+ }
+
+ /**
+ * Ensure no NPE results from passing a HEARTBEAT type message with a null {@link DagActionValue}
+ */
+ @Test
+ public void testProcessMessageWithHeartbeat() {
+ Kafka09ConsumerClient.Kafka09ConsumerRecord consumerRecord =
+ wrapDagActionStoreChangeEvent(OperationType.HEARTBEAT, "", "", "", null);
+ mockDagActionStoreChangeMonitor.processMessage(consumerRecord);
+ }
+
+ /**
+ * Tests process message with an INSERT type message
+ */
+ @Test
+ public void testProcessMessageWithInsert() {
+ Kafka09ConsumerClient.Kafka09ConsumerRecord consumerRecord =
+ wrapDagActionStoreChangeEvent(OperationType.INSERT, FLOW_GROUP, FLOW_NAME, FLOW_EXECUTION_ID, DagActionValue.LAUNCH);
+ mockDagActionStoreChangeMonitor.processMessage(consumerRecord);
Review Comment:
Added a few more verifications that checks which mocked classes are called with `processMessage`. The method mostly parses the information and passes it only `DagManager` for` kill, resume, or launch` so verified which methods are called.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#issuecomment-1789766721
Checks running on my fork: https://github.com/umustafi/gobblin/pull/8/checks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#issuecomment-1791695071
## [Codecov](https://app.codecov.io/gh/apache/gobblin/pull/3817?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#3817](https://app.codecov.io/gh/apache/gobblin/pull/3817?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (25bd334) into [master](https://app.codecov.io/gh/apache/gobblin/commit/da6b1dfd02400b6aafe5a650a6fb8d9388cdac0f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (da6b1df) will **decrease** coverage by `1.63%`.
> Report is 4 commits behind head on master.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #3817 +/- ##
============================================
- Coverage 47.63% 46.00% -1.63%
+ Complexity 11047 2181 -8866
============================================
Files 2155 416 -1739
Lines 85322 17984 -67338
Branches 9488 2194 -7294
============================================
- Hits 40645 8274 -32371
+ Misses 40984 8827 -32157
+ Partials 3693 883 -2810
```
[see 1744 files with indirect coverage changes](https://app.codecov.io/gh/apache/gobblin/pull/3817/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "Will-Lo (via GitHub)" <gi...@apache.org>.
Will-Lo commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380746566
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
Review Comment:
+1
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
Review Comment:
Since we are mocking the dagManager and orchestrator, it is possible for us to actually test side effects after each processMessage by measuring the function that is called and asserting on it. See https://stackoverflow.com/questions/9841623/mockito-how-to-verify-method-was-called-on-an-object-created-within-a-method
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#issuecomment-1792900847
`ElasticsearchTransportClientWriterTest` appears flaky, passes locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1379405563
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
Review Comment:
Should we test a negative scenario where Kafka message is too big. I have seen that error often. We need to decide what to do with those messages and handle them gracefully.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380828823
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
Review Comment:
Added verification calls to each test case to ensure the right method is called depending on the input.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "arjun4084346 (via GitHub)" <gi...@apache.org>.
arjun4084346 commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1379406292
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
Review Comment:
Can we have a test that creates records with undesired format or some other kind of bad data ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380801352
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
+ }
+
+ @Override
+ protected void startUp() {
+ super.startUp();
+ }
+ }
+
+ MockDagActionStoreChangeMonitor createMockDagActionStoreChangeMonitor() {
+ Config config = ConfigFactory.empty().withValue(ConfigurationKeys.KAFKA_BROKERS, ConfigValueFactory.fromAnyRef("localhost:0000"))
+ .withValue(Kafka09ConsumerClient.GOBBLIN_CONFIG_VALUE_DESERIALIZER_CLASS_KEY, ConfigValueFactory.fromAnyRef("org.apache.kafka.common.serialization.ByteArrayDeserializer"))
+ .withValue(ConfigurationKeys.STATE_STORE_ROOT_DIR_KEY, ConfigValueFactory.fromAnyRef("/tmp/fakeStateStore"))
+ .withValue("zookeeper.connect", ConfigValueFactory.fromAnyRef("localhost:2121"));
+ return new MockDagActionStoreChangeMonitor("dummyTopic", config, 5, true);
+ }
+
+ @BeforeClass
+ public void setup() {
+ mockDagActionStoreChangeMonitor = createMockDagActionStoreChangeMonitor();
+ mockDagActionStoreChangeMonitor.startUp();
+ }
+
+ /**
+ * Ensure no NPE results from passing a HEARTBEAT type message with a null {@link DagActionValue}
+ */
+ @Test
+ public void testProcessMessageWithHeartbeat() {
+ Kafka09ConsumerClient.Kafka09ConsumerRecord consumerRecord =
+ wrapDagActionStoreChangeEvent(OperationType.HEARTBEAT, "", "", "", null);
+ mockDagActionStoreChangeMonitor.processMessage(consumerRecord);
+ }
+
+ /**
+ * Tests process message with an INSERT type message
+ */
+ @Test
+ public void testProcessMessageWithInsert() {
+ Kafka09ConsumerClient.Kafka09ConsumerRecord consumerRecord =
+ wrapDagActionStoreChangeEvent(OperationType.INSERT, FLOW_GROUP, FLOW_NAME, FLOW_EXECUTION_ID, DagActionValue.LAUNCH);
+ mockDagActionStoreChangeMonitor.processMessage(consumerRecord);
Review Comment:
I want to verify more state ie: check meter counts but for simplicity mocked the `metric context, DagManager` etc.... processing mostly extracts properties and hands over to the `DagManager`. I can take a look to see if we can verify more functionality but capped it due to return on LOE of developing more infrastructure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add framework and unit tests for DagActionStoreChangeMonitor [gobblin]
Posted by "umustafi (via GitHub)" <gi...@apache.org>.
umustafi commented on code in PR #3817:
URL: https://github.com/apache/gobblin/pull/3817#discussion_r1380799356
##########
gobblin-modules/gobblin-kafka-09/src/test/java/org/apache/gobblin/runtime/DagActionStoreChangeMonitorTest.java:
##########
@@ -0,0 +1,154 @@
+package org.apache.gobblin.runtime;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValueFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.kafka.client.DecodeableKafkaRecord;
+import org.apache.gobblin.kafka.client.Kafka09ConsumerClient;
+import org.apache.gobblin.runtime.api.DagActionStore;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.modules.orchestration.DagManager;
+import org.apache.gobblin.service.modules.orchestration.Orchestrator;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.DagActionStoreChangeMonitor;
+import org.apache.gobblin.service.monitoring.DagActionValue;
+import org.apache.gobblin.service.monitoring.GenericStoreChangeEvent;
+import org.apache.gobblin.service.monitoring.OperationType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Tests the main functionality of {@link DagActionStoreChangeMonitor} to process {@link DagActionStoreChangeEvent} type
+ * events stored in a {@link org.apache.gobblin.kafka.client.KafkaConsumerRecord}. The
+ * processMessage(DecodeableKafkaRecord message) function should be able to gracefully process a variety of message
+ * types, even with undesired formats, without throwing exceptions.
+ */
+@Slf4j
+public class DagActionStoreChangeMonitorTest {
+ public static final String TOPIC = DagActionStoreChangeEvent.class.getSimpleName();
+ private final int PARTITION = 1;
+ private final int OFFSET = 1;
+ private final String FLOW_GROUP = "flowGroup";
+ private final String FLOW_NAME = "flowName";
+ private final String FLOW_EXECUTION_ID = "123";
+ private MockDagActionStoreChangeMonitor mockDagActionStoreChangeMonitor;
+ private int txidCounter = 0;
+
+ class MockDagActionStoreChangeMonitor extends DagActionStoreChangeMonitor {
+
+ public MockDagActionStoreChangeMonitor(String topic, Config config, int numThreads,
+ boolean isMultiActiveSchedulerEnabled) {
+ super(topic, config, mock(DagActionStore.class), mock(DagManager.class), numThreads, mock(FlowCatalog.class),
+ mock(Orchestrator.class), isMultiActiveSchedulerEnabled);
+ }
+
+ @Override
+ protected void processMessage(DecodeableKafkaRecord record) {
+ super.processMessage(record);
Review Comment:
The methods are protected access but these are in a diff module so override it only to access the method. They are in this module to use the Kafka 09 version
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@gobblin.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org