You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2022/09/01 22:54:23 UTC
[GitHub] [gobblin] ZihanLi58 opened a new pull request, #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
ZihanLi58 opened a new pull request, #3549:
URL: https://github.com/apache/gobblin/pull/3549
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
- [ ] 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-1697
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
1. Have a separate resource handler to handle all flow config request regardless it's leader or not
2. Have a new mysql spec store that support update with version
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
Unit test
### Commits
- [ ] 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
3. Subject is limited to 50 characters
4. Subject does not end with a period
5. Subject uses the imperative mood ("add", not "adding")
6. Body wraps at 72 characters
7. 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
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r974787645
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/SpecStore.java:
##########
@@ -78,6 +78,16 @@ public interface SpecStore {
*/
Spec updateSpec(Spec spec) throws IOException, SpecNotFoundException;
+ /***
+ * Update {@link Spec} in the {@link SpecStore} when current version is smaller than {@link version}.
+ * @param spec {@link Spec} to be updated.
+ * @param version largest version that current spec should be
+ * @throws IOException Exception in updating the {@link Spec}.
+ * @return Updated {@link Spec}.
+ * @throws SpecNotFoundException If {@link Spec} being updated is not present in store.
+ */
+ default Spec updateSpec(Spec spec, long version) throws IOException, SpecNotFoundException {return updateSpec(spec);};
Review Comment:
So this comment is on the outdated code, so it may not be applicable. Are we trying to change our code to handle two concurrent requests? I think we can achieve this with the right mysql commands only.
--
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
[GitHub] [gobblin] ZihanLi58 commented on pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#issuecomment-1251711184
@arjun4084346 here is the answer
1) The normal mysqlSpecStore treat update the same as add operation, and the add statement accepts add when key is existed. We used only have one host talk to mysql so this is durable. But when we enable warm standby, multi host can access mysql for the same entry at the same time. I want to fail one of the request when such thing happens to make sure they are not over write with each other. This require a new mysql catalog where we treat update and add different. Add will fail when the key already exist, update will fail when it try to update one non-existed key or the key is recently updated after we receive the request (where concurrent write happens)
2) new resources handler is specifically for warm standby where we do not care whether it's leader/follower, and directly compile and update the flow catalog. Also it needs to call update method for mysql catalog when it's a update request
3) those two are both for warmstandby, and as new resource handler needs to handle the race condition, we introduce the new mysql catalog to handle that
--
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
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r970233775
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java:
##########
@@ -364,6 +364,64 @@ public Spec getSpecWrapper(URI uri) {
* @return a map of listeners and their {@link AddSpecResponse}s
*/
public Map<String, AddSpecResponse> put(Spec spec, boolean triggerListener) throws Throwable {
+ Map<String, AddSpecResponse> responseMap = updateHelper(spec, triggerListener);
+ AddSpecResponse<String> compileResponse = responseMap.get(ServiceConfigKeys.COMPILATION_RESPONSE);
+ // Check that the flow configuration is valid and matches to a corresponding edge
+ if (isCompileSuccessful(compileResponse.getValue())) {
+ FlowSpec flowSpec = (FlowSpec) spec;
+ Object syncObject = getSyncObject(flowSpec.getUri().toString());
+ synchronized (syncObject) {
+ try {
+ if (!flowSpec.isExplain()) {
+ long startTime = System.currentTimeMillis();
+ specStore.addSpec(spec);
+ metrics.updatePutSpecTime(startTime);
+ }
+ responseMap.put(ServiceConfigKeys.COMPILATION_SUCCESSFUL, new AddSpecResponse<>("true"));
+ } catch (IOException e) {
+ throw new RuntimeException("Cannot add Spec to Spec store: " + flowSpec, e);
+ } finally {
+ syncObject.notifyAll();
+ this.specSyncObjects.remove(flowSpec.getUri().toString());
+ }
+ }
+ } else {
+ responseMap.put(ServiceConfigKeys.COMPILATION_SUCCESSFUL, new AddSpecResponse<>("false"));
+ }
+
+ return responseMap;
+ }
Review Comment:
This seems very similar to update besides that it calls specStore.addSpec(spec) vs specStore.updateSpec(spec, version). Is it possible for us to parameterize this or move logic around? e.g. `validateFlowCompilation(Flowspec spec, Callable<Void> specStoreCallable)` and send in a as a parameter. It's only used twice though so current still works, but this helps with repetition.
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_catalog/FlowCatalog.java:
##########
@@ -364,6 +364,64 @@ public Spec getSpecWrapper(URI uri) {
* @return a map of listeners and their {@link AddSpecResponse}s
*/
public Map<String, AddSpecResponse> put(Spec spec, boolean triggerListener) throws Throwable {
+ Map<String, AddSpecResponse> responseMap = updateHelper(spec, triggerListener);
+ AddSpecResponse<String> compileResponse = responseMap.get(ServiceConfigKeys.COMPILATION_RESPONSE);
+ // Check that the flow configuration is valid and matches to a corresponding edge
+ if (isCompileSuccessful(compileResponse.getValue())) {
+ FlowSpec flowSpec = (FlowSpec) spec;
+ Object syncObject = getSyncObject(flowSpec.getUri().toString());
+ synchronized (syncObject) {
+ try {
+ if (!flowSpec.isExplain()) {
+ long startTime = System.currentTimeMillis();
+ specStore.addSpec(spec);
+ metrics.updatePutSpecTime(startTime);
+ }
+ responseMap.put(ServiceConfigKeys.COMPILATION_SUCCESSFUL, new AddSpecResponse<>("true"));
+ } catch (IOException e) {
+ throw new RuntimeException("Cannot add Spec to Spec store: " + flowSpec, e);
+ } finally {
+ syncObject.notifyAll();
+ this.specSyncObjects.remove(flowSpec.getUri().toString());
+ }
+ }
+ } else {
+ responseMap.put(ServiceConfigKeys.COMPILATION_SUCCESSFUL, new AddSpecResponse<>("false"));
+ }
+
+ return responseMap;
+ }
+
+ public Map<String, AddSpecResponse> update(Spec spec, boolean triggerListener, long version) throws Throwable {
+ Map<String, AddSpecResponse> responseMap = updateHelper(spec, triggerListener);
+ AddSpecResponse<String> compileResponse = responseMap.get(ServiceConfigKeys.COMPILATION_RESPONSE);
+ // Check that the flow configuration is valid and matches to a corresponding edge
+ if (isCompileSuccessful(compileResponse.getValue())) {
+ FlowSpec flowSpec = (FlowSpec) spec;
+ Object syncObject = getSyncObject(flowSpec.getUri().toString());
+ synchronized (syncObject) {
+ try {
+ if (!flowSpec.isExplain()) {
+ long startTime = System.currentTimeMillis();
+ specStore.updateSpec(spec, version);
+ metrics.updatePutSpecTime(startTime);
+ }
+ responseMap.put(ServiceConfigKeys.COMPILATION_SUCCESSFUL, new AddSpecResponse<>("true"));
+ } catch (IOException e) {
+ throw new RuntimeException("Cannot add Spec to Spec store: " + flowSpec, e);
+ } finally {
+ syncObject.notifyAll();
+ this.specSyncObjects.remove(flowSpec.getUri().toString());
+ }
+ }
+ } else {
+ responseMap.put(ServiceConfigKeys.COMPILATION_SUCCESSFUL, new AddSpecResponse<>("false"));
+ }
+
+ return responseMap;
+ }
+
+ private Map<String, AddSpecResponse> updateHelper(Spec spec, boolean triggerListener) throws Throwable {
Review Comment:
Maybe more direct naming as to what this is doing? e.g. `getListenerResponses` and add a javadoc as to which listeners its currently listening on.
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/SpecStore.java:
##########
@@ -78,6 +78,16 @@ public interface SpecStore {
*/
Spec updateSpec(Spec spec) throws IOException, SpecNotFoundException;
+ /***
+ * Update {@link Spec} in the {@link SpecStore} when current version is smaller than {@link version}.
+ * @param spec {@link Spec} to be updated.
+ * @param version largest version that current spec should be
+ * @throws IOException Exception in updating the {@link Spec}.
+ * @return Updated {@link Spec}.
+ * @throws SpecNotFoundException If {@link Spec} being updated is not present in store.
+ */
+ default Spec updateSpec(Spec spec, long version) throws IOException, SpecNotFoundException {return updateSpec(spec);};
Review Comment:
I'm confused about the purpose of the version parameter here. What's the intention of it if we set it as the max integer?
--
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
[GitHub] [gobblin] arjun4084346 commented on pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#issuecomment-1253117667
If the MysqlSpecStore is changed to act like a MysqlSpecStoreWithUpdate , will it break anything?
--
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
[GitHub] [gobblin] ZihanLi58 merged pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 merged PR #3549:
URL: https://github.com/apache/gobblin/pull/3549
--
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
[GitHub] [gobblin] arjun4084346 commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r974784033
##########
gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowConfigResourceLocalHandler.java:
##########
@@ -161,10 +161,14 @@ public CreateResponse createFlowConfig(FlowConfig flowConfig) throws FlowConfigL
return this.createFlowConfig(flowConfig, true);
}
+ public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener) {
+ // Set the max version to be the largest value so that we blindly update the flow spec in this case
+ return updateFlowConfig(flowId, flowConfig, triggerListener, Long.MAX_VALUE);
+ }
/**
* Update flowConfig locally and trigger all listeners iff @param triggerListener is set to true
*/
- public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener) {
+ public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener, long modifiedWatermark) {
Review Comment:
Is version and watermark are the same thing?
--
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
[GitHub] [gobblin] ZihanLi58 commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r971361796
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/SpecStore.java:
##########
@@ -78,6 +78,16 @@ public interface SpecStore {
*/
Spec updateSpec(Spec spec) throws IOException, SpecNotFoundException;
+ /***
+ * Update {@link Spec} in the {@link SpecStore} when current version is smaller than {@link version}.
+ * @param spec {@link Spec} to be updated.
+ * @param version largest version that current spec should be
+ * @throws IOException Exception in updating the {@link Spec}.
+ * @return Updated {@link Spec}.
+ * @throws SpecNotFoundException If {@link Spec} being updated is not present in store.
+ */
+ default Spec updateSpec(Spec spec, long version) throws IOException, SpecNotFoundException {return updateSpec(spec);};
Review Comment:
If we are able to accept request from multi host, it's possible that two user try to modify same flow at the same time, especially for patch update, this version is introduced to avoid this situation, only the first update will go through, the second one will fail with the concurrent update exception.
--
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
[GitHub] [gobblin] Will-Lo commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
Will-Lo commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r974767122
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/api/SpecStore.java:
##########
@@ -78,6 +78,16 @@ public interface SpecStore {
*/
Spec updateSpec(Spec spec) throws IOException, SpecNotFoundException;
+ /***
+ * Update {@link Spec} in the {@link SpecStore} when current version is smaller than {@link version}.
+ * @param spec {@link Spec} to be updated.
+ * @param version largest version that current spec should be
+ * @throws IOException Exception in updating the {@link Spec}.
+ * @return Updated {@link Spec}.
+ * @throws SpecNotFoundException If {@link Spec} being updated is not present in store.
+ */
+ default Spec updateSpec(Spec spec, long version) throws IOException, SpecNotFoundException {return updateSpec(spec);};
Review Comment:
For that purpose, I think version isn't the best naming for this functionality, maybe something more related to the modifiedTimestamp? e.g. modifiedWatermark? Version makes me think that it would be directly incremental.
Reason being is that there was some interest in adding versioning to the FlowSpecStore, if this is needed for users in the future for them to undo changes, then it would make sense we refer to this timestamp differently.
--
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
[GitHub] [gobblin] ZihanLi58 commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r974792579
##########
gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowConfigResourceLocalHandler.java:
##########
@@ -161,10 +161,14 @@ public CreateResponse createFlowConfig(FlowConfig flowConfig) throws FlowConfigL
return this.createFlowConfig(flowConfig, true);
}
+ public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener) {
+ // Set the max version to be the largest value so that we blindly update the flow spec in this case
+ return updateFlowConfig(flowId, flowConfig, triggerListener, Long.MAX_VALUE);
+ }
/**
* Update flowConfig locally and trigger all listeners iff @param triggerListener is set to true
*/
- public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener) {
+ public UpdateResponse updateFlowConfig(FlowId flowId, FlowConfig flowConfig, boolean triggerListener, long modifiedWatermark) {
Review Comment:
Same thing in this PR, just update the name to avoid confusion. Basically we will not update the entry if the modification time is larger than the "modifiedWatermark" / "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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
umustafi commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r967547782
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_store/MysqlSpecStoreWithUpdate.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.spec_store;
+
+import com.google.common.base.Charsets;
+import com.typesafe.config.Config;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecSerDe;
+
+
+public class MysqlSpecStoreWithUpdate extends MysqlSpecStore{
+ // In this case, when we try to insert but key is existed, we will throw exception
+ protected static final String INSERT_STATEMENT_WITHOUT_UPDATE = "INSERT INTO %s (spec_uri, flow_group, flow_name, template_uri, "
+ + "user_to_proxy, source_identifier, destination_identifier, schedule, tag, isRunImmediately, owning_group, spec, spec_json) "
+ + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ public MysqlSpecStoreWithUpdate(Config config, SpecSerDe specSerDe) throws IOException {
+ super(config, specSerDe);
+ }
+
+ /** Bundle all changes following from schema differences against the base class. */
+ protected class SpecificSqlStatementsWithUpdate extends SpecificSqlStatements {
+ public void completeUpdatePreparedStatement(PreparedStatement statement, Spec spec, long version) throws
+ SQLException {
+ FlowSpec flowSpec = (FlowSpec) spec;
+ URI specUri = flowSpec.getUri();
+
+ int i = 0;
+
+ statement.setBlob(++i, new ByteArrayInputStream(MysqlSpecStoreWithUpdate.this.specSerDe.serialize(flowSpec)));
+ statement.setString(++i, new String(MysqlSpecStoreWithUpdate.this.specSerDe.serialize(flowSpec), Charsets.UTF_8));
+ statement.setString(++i, specUri.toString());
+ statement.setLong(++i, version);
+ }
+
+ @Override
+ protected String getTablelessInsertStatement() { return INSERT_STATEMENT_WITHOUT_UPDATE; }
+ }
+
+ @Override
+ protected SqlStatements createSqlStatements() {
+ return new SpecificSqlStatementsWithUpdate();
+ }
+
+ @Override
+ // TODO: fix to obey the `SpecStore` contract of returning the *updated* `Spec`
+ public Spec updateSpecImpl(Spec spec) throws IOException {
+ updateSpecImpl(spec, Long.MAX_VALUE);
+ return spec;
+ }
+
+ @Override
+ // TODO: fix to obey the `SpecStore` contract of returning the *updated* `Spec`
+ public Spec updateSpecImpl(Spec spec, long version) throws IOException {
+ withPreparedStatement(this.sqlStatements.updateStatement, statement -> {
+ ((SpecificSqlStatementsWithUpdate)this.sqlStatements).completeUpdatePreparedStatement(statement, spec, version);
Review Comment:
is the last value supposed to be `timestamp` or `version`? In the insert statement it looks like it is expecting a `timestamp`, maybe we can make name consistent or add a comment that says timestamp is also acting as versioning.
##########
gobblin-runtime/src/test/java/org/apache/gobblin/runtime/spec_store/MysqlSpecStoreWithUpdateTest.java:
##########
@@ -0,0 +1,399 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.spec_store;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Iterators;
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.FlowSpecSearchObject;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecSerDe;
+import org.apache.gobblin.runtime.api.SpecSerDeException;
+import org.apache.gobblin.runtime.spec_serde.GsonFlowSpecSerDe;
+import org.apache.gobblin.service.FlowId;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY;
+import static org.apache.gobblin.service.ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY;
+
+
+public class MysqlSpecStoreWithUpdateTest {
+ private static final String USER = "testUser";
+ private static final String PASSWORD = "testPassword";
+ private static final String TABLE = "spec_store";
+
+ private MysqlSpecStoreWithUpdate specStore;
+ private MysqlSpecStore oldSpecStore;
+ private final URI uri1 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg1").setFlowGroup("fn1"));
+ private final URI uri2 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg2").setFlowGroup("fn2"));
+ private final URI uri3 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg3").setFlowGroup("fn3"));
+ private final URI uri4 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg4").setFlowGroup("fn4"));
+ private FlowSpec flowSpec1, flowSpec2, flowSpec3, flowSpec4, flowSpec4_update;
+
+ public MysqlSpecStoreWithUpdateTest()
+ throws URISyntaxException { // (based on `uri1` and other initializations just above)
+ }
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testDb = TestMetastoreDatabaseFactory.get();
+
+ Config config = ConfigBuilder.create()
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, USER)
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, PASSWORD)
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, TABLE)
+ .build();
+
+ this.specStore = new MysqlSpecStoreWithUpdate(config, new TestSpecSerDe());
+ this.oldSpecStore = new OldSpecStore(config, new TestSpecSerDe());
+
+ flowSpec1 = FlowSpec.builder(this.uri1)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive("key", "value")
+ .addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive("config.with.dot", "value4")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg1")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn1").build())
+ .withDescription("Test flow spec")
+ .withVersion("Test version")
+ .build();
+ flowSpec2 = FlowSpec.builder(this.uri2)
+ .withConfig(ConfigBuilder.create().addPrimitive("converter", "value1,value2,value3")
+ .addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg2")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn2").build())
+ .withDescription("Test flow spec 2")
+ .withVersion("Test version 2")
+ .build();
+ flowSpec3 = FlowSpec.builder(this.uri3)
+ .withConfig(ConfigBuilder.create().addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg3")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn3").build())
+ .withDescription("Test flow spec 3")
+ .withVersion("Test version 3")
+ .build();
+
+ flowSpec4 = FlowSpec.builder(this.uri4)
+ .withConfig(ConfigBuilder.create().addPrimitive("key4", "value4")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg4")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn4")
+ .addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, "owningGroup4").build())
+ .withDescription("Test flow spec 4")
+ .withVersion("Test version 4")
+ .build();
+ flowSpec4_update = FlowSpec.builder(this.uri4)
+ .withConfig(ConfigBuilder.create().addPrimitive("key4", "value4_update")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg4")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn4")
+ .addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, "owningGroup4").build())
+ .withDescription("Test flow spec 4")
+ .withVersion("Test version 4")
+ .build();
+ }
+
+ @Test(expectedExceptions = IOException.class)
+ public void testSpecSearch() throws Exception {
+ // empty FlowSpecSearchObject should throw an error
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().build();
+ flowSpecSearchObject.augmentBaseGetStatement("SELECT * FROM Dummy WHERE ");
+ }
+
+ @Test
+ public void testAddSpec() throws Exception {
+ this.specStore.addSpec(this.flowSpec1);
+ this.specStore.addSpec(this.flowSpec2);
+ this.specStore.addSpec(this.flowSpec4);
+
+ Assert.assertEquals(this.specStore.getSize(), 3);
+ Assert.assertTrue(this.specStore.exists(this.uri1));
+ Assert.assertTrue(this.specStore.exists(this.uri2));
+ Assert.assertTrue(this.specStore.exists(this.uri4));
+ Assert.expectThrows(Exception.class, () -> this.specStore.addSpec(this.flowSpec1));
+ Assert.assertFalse(this.specStore.exists(URI.create("dummy")));
+ }
+
+ @Test (dependsOnMethods = "testAddSpec")
+ public void testGetSpec() throws Exception {
+ FlowSpec result = (FlowSpec) this.specStore.getSpec(this.uri1);
+ Assert.assertEquals(result, this.flowSpec1);
+
+ Collection<Spec> specs = this.specStore.getSpecs();
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ Iterator<URI> uris = this.specStore.getSpecURIs();
+ Assert.assertTrue(Iterators.contains(uris, this.uri1));
+ Assert.assertTrue(Iterators.contains(uris, this.uri2));
+
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().flowGroup("fg1").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().flowName("fn2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().flowName("fg1").flowGroup("fn2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 0);
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("key=value").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("converter=value2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("key3").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 2);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("config.with.dot=value4").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().owningGroup("owningGroup4").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetSpecWithTag() throws Exception {
+
+ //Creating and inserting flowspecs with tags
+ URI uri5 = URI.create("flowspec5");
+ FlowSpec flowSpec5 = FlowSpec.builder(uri5)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg5")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn5")
+ .addPrimitive("key5", "value5").build())
+ .withDescription("Test flow spec 5")
+ .withVersion("Test version 5")
+ .build();
+
+ URI uri6 = URI.create("flowspec6");
+ FlowSpec flowSpec6 = FlowSpec.builder(uri6)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg6")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn6")
+ .addPrimitive("key6", "value6").build())
+ .withDescription("Test flow spec 6")
+ .withVersion("Test version 6")
+ .build();
+
+ this.specStore.addSpec(flowSpec5, "dr");
+ this.specStore.addSpec(flowSpec6, "dr");
+
+ Assert.assertTrue(this.specStore.exists(uri5));
+ Assert.assertTrue(this.specStore.exists(uri6));
+ List<URI> result = new ArrayList<>();
+ this.specStore.getSpecURIsWithTag("dr").forEachRemaining(result::add);
+ Assert.assertEquals(result.size(), 2);
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetFilterSpecPaginate() throws Exception {
+ /**
+ * Sorted order of the specStore configurations is flowSpec1, flowSpec2.
+ * flowSpec3 is not included as it is a 'corrupted' flowspec
+ * flowSpec4 is not included as it doesn't have the 'filter.this.flow' property
+ * Start is the offset of the first configuration to return
+ * Count is the total number of configurations to return
+ * PropertyFilter is the property to filter by
+ */
+
+ // Start of 0 and count of 1 means start from index 0, and return one configuration only
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().start(0).count(1).propertyFilter("filter.this.flow").build();
+ Collection<Spec> specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertFalse(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+
+ // Start of 1 and count of 1 means start from index 1, and return one configuration only
+ flowSpecSearchObject = FlowSpecSearchObject.builder().start(1).count(1).propertyFilter("filter.this.flow").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertFalse(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+
+ /**
+ * Start of 0 and count of 5 means start from index 0, and return five configuration only
+ * Total of 3 flowSpecs in the DB, but flowSpec4 doesn't have 'filter.this.flow' filter so only returns 2 flowSpecs
+ * flowSpec1 and flowSpec2 match all the criteria
+ */
+ flowSpecSearchObject = FlowSpecSearchObject.builder().start(0).count(5).propertyFilter("filter.this.flow").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 2);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testUpdate() throws Exception{
+ long version = System.currentTimeMillis() /1000;
+ this.specStore.updateSpec(this.flowSpec4_update);
+ Assert.assertEquals(((FlowSpec) this.specStore.getSpec(this.uri4)), flowSpec4_update);
+ Assert.expectThrows(IOException.class, () -> this.specStore.updateSpec(flowSpec4, version));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetAllSpecPaginate() throws Exception {
+ /**
+ * Sorted order of the specStore configurations is flowSpec1, flowSpec2, flowSpec4
+ */
+ // Return all flowSpecs from index 0 to 9. Total of 3 flowSpecs only so return all 3 flowSpecs
+ Collection<Spec> specs = this.specStore.getSpecs(0,10);
+ for (Spec spec: specs) {
+ System.out.println("test" + spec.getUri());
+ }
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+
+ // Return all flowSpecs using the default get all specs function. Testing default functionality of returning everything
+ specs = this.specStore.getSpecs();
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+
+ // Return all flowSpecs from index 0 to 2 - 1. Total of 3 flowSpecs, only return first two.
Review Comment:
the index specification is a bit confusing here, are you sayign `2-1` because the end is not inclusive?
--
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
[GitHub] [gobblin] ZihanLi58 commented on pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#issuecomment-1253149496
>
@arjun4084346 I don't think that will break anything, but it will require some refractor on the old API and I do see we had assumption in a lot places that update is same as add. For me, adding this new class does not introduce any duplicity (as we expand the existing class), also it helps keep the new feature separate from the old API, and it will be easy for us to roll back via config if there is any issue when we enable the change
--
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
[GitHub] [gobblin] codecov-commenter commented on pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#issuecomment-1234939381
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3549?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#3549](https://codecov.io/gh/apache/gobblin/pull/3549?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (56c110b) into [master](https://codecov.io/gh/apache/gobblin/commit/4a80b2adc90e7f3962f85fec93aec1a51e58c818?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (4a80b2a) will **increase** coverage by `2.52%`.
> The diff coverage is `43.95%`.
```diff
@@ Coverage Diff @@
## master #3549 +/- ##
============================================
+ Coverage 46.64% 49.17% +2.52%
+ Complexity 10468 9010 -1458
============================================
Files 2087 1709 -378
Lines 81713 66215 -15498
Branches 9114 7580 -1534
============================================
- Hits 38114 32560 -5554
+ Misses 40076 30583 -9493
+ Partials 3523 3072 -451
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3549?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...obblin/service/FlowConfigResourceLocalHandler.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1yZXN0bGkvZ29iYmxpbi1mbG93LWNvbmZpZy1zZXJ2aWNlL2dvYmJsaW4tZmxvdy1jb25maWctc2VydmljZS1zZXJ2ZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vc2VydmljZS9GbG93Q29uZmlnUmVzb3VyY2VMb2NhbEhhbmRsZXIuamF2YQ==) | `15.68% <0.00%> (-0.16%)` | :arrow_down: |
| [...che/gobblin/runtime/api/InstrumentedSpecStore.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0luc3RydW1lbnRlZFNwZWNTdG9yZS5qYXZh) | `79.54% <0.00%> (-3.79%)` | :arrow_down: |
| [...java/org/apache/gobblin/runtime/api/SpecStore.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL1NwZWNTdG9yZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...estli/GobblinServiceFlowConfigResourceHandler.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9yZXN0bGkvR29iYmxpblNlcnZpY2VGbG93Q29uZmlnUmVzb3VyY2VIYW5kbGVyLmphdmE=) | `45.33% <ø> (ø)` | |
| [...iceFlowConfigV2ResourceHandlerWithWarmStandby.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9yZXN0bGkvR29iYmxpblNlcnZpY2VGbG93Q29uZmlnVjJSZXNvdXJjZUhhbmRsZXJXaXRoV2FybVN0YW5kYnkuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ache/gobblin/runtime/spec\_catalog/FlowCatalog.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvc3BlY19jYXRhbG9nL0Zsb3dDYXRhbG9nLmphdmE=) | `44.44% <42.10%> (-2.66%)` | :arrow_down: |
| [...gobblin/runtime/spec\_store/MysqlBaseSpecStore.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvc3BlY19zdG9yZS9NeXNxbEJhc2VTcGVjU3RvcmUuamF2YQ==) | `89.07% <100.00%> (+0.18%)` | :arrow_up: |
| [...n/runtime/spec\_store/MysqlSpecStoreWithUpdate.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvc3BlY19zdG9yZS9NeXNxbFNwZWNTdG9yZVdpdGhVcGRhdGUuamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [...a/org/apache/gobblin/util/limiter/NoopLimiter.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvbGltaXRlci9Ob29wTGltaXRlci5qYXZh) | `40.00% <0.00%> (-20.00%)` | :arrow_down: |
| [...he/gobblin/source/PartitionAwareFileRetriever.java](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NvdXJjZS9QYXJ0aXRpb25Bd2FyZUZpbGVSZXRyaWV2ZXIuamF2YQ==) | `48.14% <0.00%> (-7.41%)` | :arrow_down: |
| ... and [386 more](https://codecov.io/gh/apache/gobblin/pull/3549/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
: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=The+Apache+Software+Foundation)
--
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
[GitHub] [gobblin] arjun4084346 commented on pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
arjun4084346 commented on PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#issuecomment-1251706720
Please elaborate 1) why MysqlSpecStoreWithUpdate is required 2) why new resource handler is requires and 2) how both are related ?
--
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
[GitHub] [gobblin] umustafi commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
umustafi commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r974768606
##########
gobblin-runtime/src/test/java/org/apache/gobblin/runtime/spec_store/MysqlSpecStoreWithUpdateTest.java:
##########
@@ -0,0 +1,399 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.spec_store;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Iterators;
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.FlowSpecSearchObject;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecSerDe;
+import org.apache.gobblin.runtime.api.SpecSerDeException;
+import org.apache.gobblin.runtime.spec_serde.GsonFlowSpecSerDe;
+import org.apache.gobblin.service.FlowId;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY;
+import static org.apache.gobblin.service.ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY;
+
+
+public class MysqlSpecStoreWithUpdateTest {
+ private static final String USER = "testUser";
+ private static final String PASSWORD = "testPassword";
+ private static final String TABLE = "spec_store";
+
+ private MysqlSpecStoreWithUpdate specStore;
+ private MysqlSpecStore oldSpecStore;
+ private final URI uri1 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg1").setFlowGroup("fn1"));
+ private final URI uri2 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg2").setFlowGroup("fn2"));
+ private final URI uri3 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg3").setFlowGroup("fn3"));
+ private final URI uri4 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg4").setFlowGroup("fn4"));
+ private FlowSpec flowSpec1, flowSpec2, flowSpec3, flowSpec4, flowSpec4_update;
+
+ public MysqlSpecStoreWithUpdateTest()
+ throws URISyntaxException { // (based on `uri1` and other initializations just above)
+ }
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testDb = TestMetastoreDatabaseFactory.get();
+
+ Config config = ConfigBuilder.create()
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, USER)
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, PASSWORD)
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, TABLE)
+ .build();
+
+ this.specStore = new MysqlSpecStoreWithUpdate(config, new TestSpecSerDe());
+ this.oldSpecStore = new OldSpecStore(config, new TestSpecSerDe());
+
+ flowSpec1 = FlowSpec.builder(this.uri1)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive("key", "value")
+ .addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive("config.with.dot", "value4")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg1")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn1").build())
+ .withDescription("Test flow spec")
+ .withVersion("Test version")
+ .build();
+ flowSpec2 = FlowSpec.builder(this.uri2)
+ .withConfig(ConfigBuilder.create().addPrimitive("converter", "value1,value2,value3")
+ .addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg2")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn2").build())
+ .withDescription("Test flow spec 2")
+ .withVersion("Test version 2")
+ .build();
+ flowSpec3 = FlowSpec.builder(this.uri3)
+ .withConfig(ConfigBuilder.create().addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg3")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn3").build())
+ .withDescription("Test flow spec 3")
+ .withVersion("Test version 3")
+ .build();
+
+ flowSpec4 = FlowSpec.builder(this.uri4)
+ .withConfig(ConfigBuilder.create().addPrimitive("key4", "value4")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg4")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn4")
+ .addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, "owningGroup4").build())
+ .withDescription("Test flow spec 4")
+ .withVersion("Test version 4")
+ .build();
+ flowSpec4_update = FlowSpec.builder(this.uri4)
+ .withConfig(ConfigBuilder.create().addPrimitive("key4", "value4_update")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg4")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn4")
+ .addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, "owningGroup4").build())
+ .withDescription("Test flow spec 4")
+ .withVersion("Test version 4")
+ .build();
+ }
+
+ @Test(expectedExceptions = IOException.class)
+ public void testSpecSearch() throws Exception {
+ // empty FlowSpecSearchObject should throw an error
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().build();
+ flowSpecSearchObject.augmentBaseGetStatement("SELECT * FROM Dummy WHERE ");
+ }
+
+ @Test
+ public void testAddSpec() throws Exception {
+ this.specStore.addSpec(this.flowSpec1);
+ this.specStore.addSpec(this.flowSpec2);
+ this.specStore.addSpec(this.flowSpec4);
+
+ Assert.assertEquals(this.specStore.getSize(), 3);
+ Assert.assertTrue(this.specStore.exists(this.uri1));
+ Assert.assertTrue(this.specStore.exists(this.uri2));
+ Assert.assertTrue(this.specStore.exists(this.uri4));
+ Assert.expectThrows(Exception.class, () -> this.specStore.addSpec(this.flowSpec1));
+ Assert.assertFalse(this.specStore.exists(URI.create("dummy")));
+ }
+
+ @Test (dependsOnMethods = "testAddSpec")
+ public void testGetSpec() throws Exception {
+ FlowSpec result = (FlowSpec) this.specStore.getSpec(this.uri1);
+ Assert.assertEquals(result, this.flowSpec1);
+
+ Collection<Spec> specs = this.specStore.getSpecs();
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ Iterator<URI> uris = this.specStore.getSpecURIs();
+ Assert.assertTrue(Iterators.contains(uris, this.uri1));
+ Assert.assertTrue(Iterators.contains(uris, this.uri2));
+
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().flowGroup("fg1").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().flowName("fn2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().flowName("fg1").flowGroup("fn2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 0);
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("key=value").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("converter=value2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("key3").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 2);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("config.with.dot=value4").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().owningGroup("owningGroup4").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetSpecWithTag() throws Exception {
+
+ //Creating and inserting flowspecs with tags
+ URI uri5 = URI.create("flowspec5");
+ FlowSpec flowSpec5 = FlowSpec.builder(uri5)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg5")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn5")
+ .addPrimitive("key5", "value5").build())
+ .withDescription("Test flow spec 5")
+ .withVersion("Test version 5")
+ .build();
+
+ URI uri6 = URI.create("flowspec6");
+ FlowSpec flowSpec6 = FlowSpec.builder(uri6)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg6")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn6")
+ .addPrimitive("key6", "value6").build())
+ .withDescription("Test flow spec 6")
+ .withVersion("Test version 6")
+ .build();
+
+ this.specStore.addSpec(flowSpec5, "dr");
+ this.specStore.addSpec(flowSpec6, "dr");
+
+ Assert.assertTrue(this.specStore.exists(uri5));
+ Assert.assertTrue(this.specStore.exists(uri6));
+ List<URI> result = new ArrayList<>();
+ this.specStore.getSpecURIsWithTag("dr").forEachRemaining(result::add);
+ Assert.assertEquals(result.size(), 2);
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetFilterSpecPaginate() throws Exception {
+ /**
+ * Sorted order of the specStore configurations is flowSpec1, flowSpec2.
+ * flowSpec3 is not included as it is a 'corrupted' flowspec
+ * flowSpec4 is not included as it doesn't have the 'filter.this.flow' property
+ * Start is the offset of the first configuration to return
+ * Count is the total number of configurations to return
+ * PropertyFilter is the property to filter by
+ */
+
+ // Start of 0 and count of 1 means start from index 0, and return one configuration only
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().start(0).count(1).propertyFilter("filter.this.flow").build();
+ Collection<Spec> specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertFalse(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+
+ // Start of 1 and count of 1 means start from index 1, and return one configuration only
+ flowSpecSearchObject = FlowSpecSearchObject.builder().start(1).count(1).propertyFilter("filter.this.flow").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertFalse(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+
+ /**
+ * Start of 0 and count of 5 means start from index 0, and return five configuration only
+ * Total of 3 flowSpecs in the DB, but flowSpec4 doesn't have 'filter.this.flow' filter so only returns 2 flowSpecs
+ * flowSpec1 and flowSpec2 match all the criteria
+ */
+ flowSpecSearchObject = FlowSpecSearchObject.builder().start(0).count(5).propertyFilter("filter.this.flow").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 2);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testUpdate() throws Exception{
+ long version = System.currentTimeMillis() /1000;
+ this.specStore.updateSpec(this.flowSpec4_update);
+ Assert.assertEquals(((FlowSpec) this.specStore.getSpec(this.uri4)), flowSpec4_update);
+ Assert.expectThrows(IOException.class, () -> this.specStore.updateSpec(flowSpec4, version));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetAllSpecPaginate() throws Exception {
+ /**
+ * Sorted order of the specStore configurations is flowSpec1, flowSpec2, flowSpec4
+ */
+ // Return all flowSpecs from index 0 to 9. Total of 3 flowSpecs only so return all 3 flowSpecs
+ Collection<Spec> specs = this.specStore.getSpecs(0,10);
+ for (Spec spec: specs) {
+ System.out.println("test" + spec.getUri());
+ }
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+
+ // Return all flowSpecs using the default get all specs function. Testing default functionality of returning everything
+ specs = this.specStore.getSpecs();
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+
+ // Return all flowSpecs from index 0 to 2 - 1. Total of 3 flowSpecs, only return first two.
Review Comment:
oh I see, do you want to say [0, 2) inclusive, exclusive notation
--
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
[GitHub] [gobblin] ZihanLi58 commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r969986709
##########
gobblin-runtime/src/test/java/org/apache/gobblin/runtime/spec_store/MysqlSpecStoreWithUpdateTest.java:
##########
@@ -0,0 +1,399 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.spec_store;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.lang3.ArrayUtils;
+import org.testng.Assert;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.collect.Iterators;
+import com.typesafe.config.Config;
+
+import org.apache.gobblin.config.ConfigBuilder;
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.FlowSpecSearchObject;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecSerDe;
+import org.apache.gobblin.runtime.api.SpecSerDeException;
+import org.apache.gobblin.runtime.spec_serde.GsonFlowSpecSerDe;
+import org.apache.gobblin.service.FlowId;
+
+import static org.apache.gobblin.service.ServiceConfigKeys.FLOW_DESTINATION_IDENTIFIER_KEY;
+import static org.apache.gobblin.service.ServiceConfigKeys.FLOW_SOURCE_IDENTIFIER_KEY;
+
+
+public class MysqlSpecStoreWithUpdateTest {
+ private static final String USER = "testUser";
+ private static final String PASSWORD = "testPassword";
+ private static final String TABLE = "spec_store";
+
+ private MysqlSpecStoreWithUpdate specStore;
+ private MysqlSpecStore oldSpecStore;
+ private final URI uri1 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg1").setFlowGroup("fn1"));
+ private final URI uri2 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg2").setFlowGroup("fn2"));
+ private final URI uri3 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg3").setFlowGroup("fn3"));
+ private final URI uri4 = FlowSpec.Utils.createFlowSpecUri(new FlowId().setFlowName("fg4").setFlowGroup("fn4"));
+ private FlowSpec flowSpec1, flowSpec2, flowSpec3, flowSpec4, flowSpec4_update;
+
+ public MysqlSpecStoreWithUpdateTest()
+ throws URISyntaxException { // (based on `uri1` and other initializations just above)
+ }
+
+ @BeforeClass
+ public void setUp() throws Exception {
+ ITestMetastoreDatabase testDb = TestMetastoreDatabaseFactory.get();
+
+ Config config = ConfigBuilder.create()
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_URL_KEY, testDb.getJdbcUrl())
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_USER_KEY, USER)
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY, PASSWORD)
+ .addPrimitive(ConfigurationKeys.STATE_STORE_DB_TABLE_KEY, TABLE)
+ .build();
+
+ this.specStore = new MysqlSpecStoreWithUpdate(config, new TestSpecSerDe());
+ this.oldSpecStore = new OldSpecStore(config, new TestSpecSerDe());
+
+ flowSpec1 = FlowSpec.builder(this.uri1)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive("key", "value")
+ .addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive("config.with.dot", "value4")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg1")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn1").build())
+ .withDescription("Test flow spec")
+ .withVersion("Test version")
+ .build();
+ flowSpec2 = FlowSpec.builder(this.uri2)
+ .withConfig(ConfigBuilder.create().addPrimitive("converter", "value1,value2,value3")
+ .addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg2")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn2").build())
+ .withDescription("Test flow spec 2")
+ .withVersion("Test version 2")
+ .build();
+ flowSpec3 = FlowSpec.builder(this.uri3)
+ .withConfig(ConfigBuilder.create().addPrimitive("key3", "value3")
+ .addPrimitive("filter.this.flow", true)
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg3")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn3").build())
+ .withDescription("Test flow spec 3")
+ .withVersion("Test version 3")
+ .build();
+
+ flowSpec4 = FlowSpec.builder(this.uri4)
+ .withConfig(ConfigBuilder.create().addPrimitive("key4", "value4")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg4")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn4")
+ .addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, "owningGroup4").build())
+ .withDescription("Test flow spec 4")
+ .withVersion("Test version 4")
+ .build();
+ flowSpec4_update = FlowSpec.builder(this.uri4)
+ .withConfig(ConfigBuilder.create().addPrimitive("key4", "value4_update")
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg4")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn4")
+ .addPrimitive(ConfigurationKeys.FLOW_OWNING_GROUP_KEY, "owningGroup4").build())
+ .withDescription("Test flow spec 4")
+ .withVersion("Test version 4")
+ .build();
+ }
+
+ @Test(expectedExceptions = IOException.class)
+ public void testSpecSearch() throws Exception {
+ // empty FlowSpecSearchObject should throw an error
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().build();
+ flowSpecSearchObject.augmentBaseGetStatement("SELECT * FROM Dummy WHERE ");
+ }
+
+ @Test
+ public void testAddSpec() throws Exception {
+ this.specStore.addSpec(this.flowSpec1);
+ this.specStore.addSpec(this.flowSpec2);
+ this.specStore.addSpec(this.flowSpec4);
+
+ Assert.assertEquals(this.specStore.getSize(), 3);
+ Assert.assertTrue(this.specStore.exists(this.uri1));
+ Assert.assertTrue(this.specStore.exists(this.uri2));
+ Assert.assertTrue(this.specStore.exists(this.uri4));
+ Assert.expectThrows(Exception.class, () -> this.specStore.addSpec(this.flowSpec1));
+ Assert.assertFalse(this.specStore.exists(URI.create("dummy")));
+ }
+
+ @Test (dependsOnMethods = "testAddSpec")
+ public void testGetSpec() throws Exception {
+ FlowSpec result = (FlowSpec) this.specStore.getSpec(this.uri1);
+ Assert.assertEquals(result, this.flowSpec1);
+
+ Collection<Spec> specs = this.specStore.getSpecs();
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ Iterator<URI> uris = this.specStore.getSpecURIs();
+ Assert.assertTrue(Iterators.contains(uris, this.uri1));
+ Assert.assertTrue(Iterators.contains(uris, this.uri2));
+
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().flowGroup("fg1").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().flowName("fn2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().flowName("fg1").flowGroup("fn2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 0);
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("key=value").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("converter=value2").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("key3").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 2);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().propertyFilter("config.with.dot=value4").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+
+ flowSpecSearchObject = FlowSpecSearchObject.builder().owningGroup("owningGroup4").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetSpecWithTag() throws Exception {
+
+ //Creating and inserting flowspecs with tags
+ URI uri5 = URI.create("flowspec5");
+ FlowSpec flowSpec5 = FlowSpec.builder(uri5)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg5")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn5")
+ .addPrimitive("key5", "value5").build())
+ .withDescription("Test flow spec 5")
+ .withVersion("Test version 5")
+ .build();
+
+ URI uri6 = URI.create("flowspec6");
+ FlowSpec flowSpec6 = FlowSpec.builder(uri6)
+ .withConfig(ConfigBuilder.create()
+ .addPrimitive(FLOW_SOURCE_IDENTIFIER_KEY, "source")
+ .addPrimitive(FLOW_DESTINATION_IDENTIFIER_KEY, "destination")
+ .addPrimitive(ConfigurationKeys.FLOW_GROUP_KEY, "fg6")
+ .addPrimitive(ConfigurationKeys.FLOW_NAME_KEY, "fn6")
+ .addPrimitive("key6", "value6").build())
+ .withDescription("Test flow spec 6")
+ .withVersion("Test version 6")
+ .build();
+
+ this.specStore.addSpec(flowSpec5, "dr");
+ this.specStore.addSpec(flowSpec6, "dr");
+
+ Assert.assertTrue(this.specStore.exists(uri5));
+ Assert.assertTrue(this.specStore.exists(uri6));
+ List<URI> result = new ArrayList<>();
+ this.specStore.getSpecURIsWithTag("dr").forEachRemaining(result::add);
+ Assert.assertEquals(result.size(), 2);
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetFilterSpecPaginate() throws Exception {
+ /**
+ * Sorted order of the specStore configurations is flowSpec1, flowSpec2.
+ * flowSpec3 is not included as it is a 'corrupted' flowspec
+ * flowSpec4 is not included as it doesn't have the 'filter.this.flow' property
+ * Start is the offset of the first configuration to return
+ * Count is the total number of configurations to return
+ * PropertyFilter is the property to filter by
+ */
+
+ // Start of 0 and count of 1 means start from index 0, and return one configuration only
+ FlowSpecSearchObject flowSpecSearchObject = FlowSpecSearchObject.builder().start(0).count(1).propertyFilter("filter.this.flow").build();
+ Collection<Spec> specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertFalse(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+
+ // Start of 1 and count of 1 means start from index 1, and return one configuration only
+ flowSpecSearchObject = FlowSpecSearchObject.builder().start(1).count(1).propertyFilter("filter.this.flow").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 1);
+ Assert.assertFalse(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+
+ /**
+ * Start of 0 and count of 5 means start from index 0, and return five configuration only
+ * Total of 3 flowSpecs in the DB, but flowSpec4 doesn't have 'filter.this.flow' filter so only returns 2 flowSpecs
+ * flowSpec1 and flowSpec2 match all the criteria
+ */
+ flowSpecSearchObject = FlowSpecSearchObject.builder().start(0).count(5).propertyFilter("filter.this.flow").build();
+ specs = this.specStore.getSpecs(flowSpecSearchObject);
+ Assert.assertEquals(specs.size(), 2);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertFalse(specs.contains(this.flowSpec4));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testUpdate() throws Exception{
+ long version = System.currentTimeMillis() /1000;
+ this.specStore.updateSpec(this.flowSpec4_update);
+ Assert.assertEquals(((FlowSpec) this.specStore.getSpec(this.uri4)), flowSpec4_update);
+ Assert.expectThrows(IOException.class, () -> this.specStore.updateSpec(flowSpec4, version));
+ }
+
+ @Test (dependsOnMethods = "testGetSpec")
+ public void testGetAllSpecPaginate() throws Exception {
+ /**
+ * Sorted order of the specStore configurations is flowSpec1, flowSpec2, flowSpec4
+ */
+ // Return all flowSpecs from index 0 to 9. Total of 3 flowSpecs only so return all 3 flowSpecs
+ Collection<Spec> specs = this.specStore.getSpecs(0,10);
+ for (Spec spec: specs) {
+ System.out.println("test" + spec.getUri());
+ }
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+
+ // Return all flowSpecs using the default get all specs function. Testing default functionality of returning everything
+ specs = this.specStore.getSpecs();
+ Assert.assertEquals(specs.size(), 3);
+ Assert.assertTrue(specs.contains(this.flowSpec1));
+ Assert.assertTrue(specs.contains(this.flowSpec2));
+ Assert.assertTrue(specs.contains(this.flowSpec4));
+
+ // Return all flowSpecs from index 0 to 2 - 1. Total of 3 flowSpecs, only return first two.
Review Comment:
Because we call this.specStore.getSpecs(0,2), which should return 2 specs from index 0, so the index should be 0 to (2-1) as we start from 0.
--
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
[GitHub] [gobblin] ZihanLi58 commented on a diff in pull request #3549: [GOBBLIN-1697]Have a separate resource handler to rely on CDC stream to do message forwarding
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3549:
URL: https://github.com/apache/gobblin/pull/3549#discussion_r969984910
##########
gobblin-runtime/src/main/java/org/apache/gobblin/runtime/spec_store/MysqlSpecStoreWithUpdate.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.runtime.spec_store;
+
+import com.google.common.base.Charsets;
+import com.typesafe.config.Config;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import org.apache.gobblin.runtime.api.FlowSpec;
+import org.apache.gobblin.runtime.api.Spec;
+import org.apache.gobblin.runtime.api.SpecSerDe;
+
+
+public class MysqlSpecStoreWithUpdate extends MysqlSpecStore{
+ // In this case, when we try to insert but key is existed, we will throw exception
+ protected static final String INSERT_STATEMENT_WITHOUT_UPDATE = "INSERT INTO %s (spec_uri, flow_group, flow_name, template_uri, "
+ + "user_to_proxy, source_identifier, destination_identifier, schedule, tag, isRunImmediately, owning_group, spec, spec_json) "
+ + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
+ public MysqlSpecStoreWithUpdate(Config config, SpecSerDe specSerDe) throws IOException {
+ super(config, specSerDe);
+ }
+
+ /** Bundle all changes following from schema differences against the base class. */
+ protected class SpecificSqlStatementsWithUpdate extends SpecificSqlStatements {
+ public void completeUpdatePreparedStatement(PreparedStatement statement, Spec spec, long version) throws
+ SQLException {
+ FlowSpec flowSpec = (FlowSpec) spec;
+ URI specUri = flowSpec.getUri();
+
+ int i = 0;
+
+ statement.setBlob(++i, new ByteArrayInputStream(MysqlSpecStoreWithUpdate.this.specSerDe.serialize(flowSpec)));
+ statement.setString(++i, new String(MysqlSpecStoreWithUpdate.this.specSerDe.serialize(flowSpec), Charsets.UTF_8));
+ statement.setString(++i, specUri.toString());
+ statement.setLong(++i, version);
+ }
+
+ @Override
+ protected String getTablelessInsertStatement() { return INSERT_STATEMENT_WITHOUT_UPDATE; }
+ }
+
+ @Override
+ protected SqlStatements createSqlStatements() {
+ return new SpecificSqlStatementsWithUpdate();
+ }
+
+ @Override
+ // TODO: fix to obey the `SpecStore` contract of returning the *updated* `Spec`
+ public Spec updateSpecImpl(Spec spec) throws IOException {
+ updateSpecImpl(spec, Long.MAX_VALUE);
+ return spec;
+ }
+
+ @Override
+ // TODO: fix to obey the `SpecStore` contract of returning the *updated* `Spec`
+ public Spec updateSpecImpl(Spec spec, long version) throws IOException {
+ withPreparedStatement(this.sqlStatements.updateStatement, statement -> {
+ ((SpecificSqlStatementsWithUpdate)this.sqlStatements).completeUpdatePreparedStatement(statement, spec, version);
Review Comment:
it's a timestamp which is acting as versioning, I added the comments to explain that
--
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