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/04/05 19:39:37 UTC
[GitHub] [gobblin] homatthew opened a new pull request, #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
homatthew opened a new pull request, #3490:
URL: https://github.com/apache/gobblin/pull/3490
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-1613] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1613
### Description
- [X] Here are some details about my PR, including screenshots (if applicable):
Add optional field to GMCE schema for specifying metadata writers that should write. I've added corresponding code to filter base on this metadata writer field the GobblinMCEWriter
### Tests
- [X] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
Unit tests added and tested on holdem using iceberg metadata repair job
### 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
[GitHub] [gobblin] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r843198564
##########
gobblin-iceberg/build.gradle:
##########
@@ -56,6 +56,7 @@ dependencies {
testCompile project(path: ':gobblin-modules:gobblin-kafka-common', configuration: 'tests')
testCompile externalDependency.testng
testCompile externalDependency.mockito
+ testCompile externalDependency.powermock
Review Comment:
Mocking static method. Need to use this to mock the GobblinConstructorUtils and the FileSystem static calls
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r843199775
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java:
##########
@@ -294,12 +298,14 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
}
// Add fault tolerant ability and make sure we can emit GTE as desired
- private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ @VisibleForTesting
Review Comment:
The write envelope method that calls this method has complicated logic related to hive that is unrelated to this change. So it makes sense to test this internal method instead of the public `writeEnvelope` 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
[GitHub] [gobblin] codecov-commenter commented on pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#issuecomment-1095434061
# [Codecov](https://codecov.io/gh/apache/gobblin/pull/3490?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 [#3490](https://codecov.io/gh/apache/gobblin/pull/3490?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (128d469) into [master](https://codecov.io/gh/apache/gobblin/commit/b26bd92f06ce3a373d5ec8fae440456654108ad1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b26bd92) will **increase** coverage by `4.09%`.
> The diff coverage is `n/a`.
```diff
@@ Coverage Diff @@
## master #3490 +/- ##
============================================
+ Coverage 46.63% 50.73% +4.09%
+ Complexity 10368 5692 -4676
============================================
Files 2076 1037 -1039
Lines 81114 39557 -41557
Branches 9055 4441 -4614
============================================
- Hits 37828 20068 -17760
+ Misses 39803 17757 -22046
+ Partials 3483 1732 -1751
```
| [Impacted Files](https://codecov.io/gh/apache/gobblin/pull/3490?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [.../apache/gobblin/runtime/api/JobExecutionState.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkV4ZWN1dGlvblN0YXRlLmphdmE=) | `79.43% <0.00%> (-0.94%)` | :arrow_down: |
| [...main/java/org/apache/gobblin/yarn/YarnService.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi15YXJuL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3lhcm4vWWFyblNlcnZpY2UuamF2YQ==) | `15.46% <0.00%> (-0.52%)` | :arrow_down: |
| [...pache/gobblin/iceberg/writer/GobblinMCEWriter.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1pY2ViZXJnL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ljZWJlcmcvd3JpdGVyL0dvYmJsaW5NQ0VXcml0ZXIuamF2YQ==) | | |
| [...gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tYXprYWJhbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9hemthYmFuL0F6a2FiYW5Hb2JibGluWWFybkFwcExhdW5jaGVyLmphdmE=) | | |
| [...obblin/util/function/CheckedExceptionFunction.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvZnVuY3Rpb24vQ2hlY2tlZEV4Y2VwdGlvbkZ1bmN0aW9uLmphdmE=) | | |
| [...obblin/service/modules/flowgraph/BaseDataNode.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1zZXJ2aWNlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NlcnZpY2UvbW9kdWxlcy9mbG93Z3JhcGgvQmFzZURhdGFOb2RlLmphdmE=) | | |
| [...in/azkaban/AzkabanGobblinLocalYarnAppLauncher.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4tYXprYWJhbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZ29iYmxpbi9hemthYmFuL0F6a2FiYW5Hb2JibGluTG9jYWxZYXJuQXBwTGF1bmNoZXIuamF2YQ==) | | |
| [...tractor/extract/kafka/KafkaStreamingExtractor.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4ta2Fma2EtY29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3NvdXJjZS9leHRyYWN0b3IvZXh0cmFjdC9rYWZrYS9LYWZrYVN0cmVhbWluZ0V4dHJhY3Rvci5qYXZh) | | |
| [.../org/apache/gobblin/cluster/GobblinTaskRunner.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvR29iYmxpblRhc2tSdW5uZXIuamF2YQ==) | | |
| [...g/apache/gobblin/aws/GobblinAWSClusterManager.java](https://codecov.io/gh/apache/gobblin/pull/3490/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-Z29iYmxpbi1hd3Mvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vYXdzL0dvYmJsaW5BV1NDbHVzdGVyTWFuYWdlci5qYXZh) | | |
| ... and [1033 more](https://codecov.io/gh/apache/gobblin/pull/3490/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) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/gobblin/pull/3490?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/gobblin/pull/3490?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [b26bd92...128d469](https://codecov.io/gh/apache/gobblin/pull/3490?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?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] ZihanLi58 merged pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
ZihanLi58 merged PR #3490:
URL: https://github.com/apache/gobblin/pull/3490
--
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 #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r844463828
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java:
##########
@@ -294,12 +298,14 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
}
// Add fault tolerant ability and make sure we can emit GTE as desired
- private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ @VisibleForTesting
+ void write(RecordEnvelope<GenericRecord> recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
boolean meetException = false;
String dbName = spec.getTable().getDbName();
String tableName = spec.getTable().getTableName();
String tableString = Joiner.on(TABLE_NAME_DELIMITER).join(dbName, tableName);
- for (MetadataWriter writer : metadataWriters) {
+ List<MetadataWriter> allowedWriters = getAllowedMetadataWriters(toGMCE(recordEnvelope), metadataWriters);
Review Comment:
Can we avoid call toGMCE again in this method? As deepCopy is actually expansive..
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriterTest.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.iceberg.writer;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.hive.HiveTable;
+import org.apache.gobblin.hive.spec.HiveSpec;
+import org.apache.gobblin.hive.writer.MetadataWriter;
+import org.apache.gobblin.metadata.DatasetIdentifier;
+import org.apache.gobblin.metadata.GobblinMetadataChangeEvent;
+import org.apache.gobblin.metadata.OperationType;
+import org.apache.gobblin.metadata.SchemaSource;
+import org.apache.gobblin.source.extractor.extract.LongWatermark;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaPartition;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaStreamingExtractor;
+import org.apache.gobblin.stream.RecordEnvelope;
+import org.apache.gobblin.util.ClustersNames;
+import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.powermock.modules.testng.PowerMockTestCase;
+import org.testng.IObjectFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Test class uses PowerMockito and Testng
+ * References:
+ * https://github.com/powermock/powermock/issues/434
+ * https://jivimberg.io/blog/2016/04/03/using-powermock-plus-testng-to-mock-a-static-class/
+ * https://www.igorkromin.net/index.php/2018/10/04/how-to-fix-powermock-exception-linkageerror-loader-constraint-violation/
+ */
+@PrepareForTest({GobblinConstructorUtils.class, FileSystem.class})
+@PowerMockIgnore("javax.management.*")
+public class GobblinMCEWriterTest extends PowerMockTestCase {
+
+ private String dbName = "hivedb";
+ private String tableName = "testTable";
+ private GobblinMCEWriter gobblinMCEWriter;
+ private KafkaStreamingExtractor.KafkaWatermark watermark;
+
+ private GobblinMetadataChangeEvent.Builder gmceBuilder;
+
+ @Mock
+ private MetadataWriter mockWriter;
+
+ @Mock
+ private FileSystem fs;
+
+ @Mock
+ private HiveSpec mockHiveSpec;
+
+ @AfterMethod
+ public void clean() throws Exception {
+ gobblinMCEWriter.close();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ gmceBuilder = GobblinMetadataChangeEvent.newBuilder()
+ .setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:li:dataPlatform:hdfs")
+ .setNativeName("/data/tracking/testTable")
Review Comment:
Change a test path and test platform as this seems specific to LinkedIn?
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriterTest.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.iceberg.writer;
+
Review Comment:
Thanks for adding this test job!
##########
gobblin-iceberg/build.gradle:
##########
@@ -56,6 +56,7 @@ dependencies {
testCompile project(path: ':gobblin-modules:gobblin-kafka-common', configuration: 'tests')
testCompile externalDependency.testng
testCompile externalDependency.mockito
+ testCompile externalDependency.powermock
Review Comment:
Can you add these comments to code as well? so it's easily for us to track and understand in the future
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r844705020
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriterTest.java:
##########
@@ -346,7 +349,10 @@ public void testFaultTolerant() throws Exception {
Mockito.doThrow(new IOException("Test failure")).when(mockWriter).writeEnvelope(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
gobblinMCEWriter.metadataWriters.add(0, mockWriter);
- GenericRecord genericGmce = GenericData.get().deepCopy(gmce.getSchema(), gmce);
+ GobblinMetadataChangeEvent gmceWithMockWriter = SpecificData.get().deepCopy(gmce.getSchema(), gmce);
+ gmceWithMockWriter.setAllowedMetadataWriters(Arrays.asList(IcebergMetadataWriter.class.getName(), mockWriter.getClass().getName()));
+
+ GenericRecord genericGmce = GenericData.get().deepCopy(gmceWithMockWriter.getSchema(), gmceWithMockWriter);
Review Comment:
This is the only substance change in the file. Add mock writer to the allow list in the GMCE
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/IcebergMetadataWriterTest.java:
##########
@@ -346,7 +349,10 @@ public void testFaultTolerant() throws Exception {
Mockito.doThrow(new IOException("Test failure")).when(mockWriter).writeEnvelope(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any());
gobblinMCEWriter.metadataWriters.add(0, mockWriter);
- GenericRecord genericGmce = GenericData.get().deepCopy(gmce.getSchema(), gmce);
+ GobblinMetadataChangeEvent gmceWithMockWriter = SpecificData.get().deepCopy(gmce.getSchema(), gmce);
+ gmceWithMockWriter.setAllowedMetadataWriters(Arrays.asList(IcebergMetadataWriter.class.getName(), mockWriter.getClass().getName()));
+
+ GenericRecord genericGmce = GenericData.get().deepCopy(gmceWithMockWriter.getSchema(), gmceWithMockWriter);
Review Comment:
This is the only substance change in the test. Add mock writer to the allow list in the GMCE
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r844706109
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriterTest.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.iceberg.writer;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
+import lombok.SneakyThrows;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.hive.HiveTable;
+import org.apache.gobblin.hive.spec.HiveSpec;
+import org.apache.gobblin.hive.writer.HiveMetadataWriter;
+import org.apache.gobblin.hive.writer.MetadataWriter;
+import org.apache.gobblin.metadata.DatasetIdentifier;
+import org.apache.gobblin.metadata.GobblinMetadataChangeEvent;
+import org.apache.gobblin.metadata.OperationType;
+import org.apache.gobblin.metadata.SchemaSource;
+import org.apache.gobblin.source.extractor.extract.LongWatermark;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaPartition;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaStreamingExtractor;
+import org.apache.gobblin.stream.RecordEnvelope;
+import org.apache.gobblin.util.ClustersNames;
+import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.powermock.modules.testng.PowerMockTestCase;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Test class uses PowerMockito and Testng
+ * References:
+ * https://github.com/powermock/powermock/issues/434
+ * https://jivimberg.io/blog/2016/04/03/using-powermock-plus-testng-to-mock-a-static-class/
+ * https://www.igorkromin.net/index.php/2018/10/04/how-to-fix-powermock-exception-linkageerror-loader-constraint-violation/
+ */
+@PrepareForTest({GobblinConstructorUtils.class, FileSystem.class})
+@PowerMockIgnore("javax.management.*")
+public class GobblinMCEWriterTest extends PowerMockTestCase {
+
+ private String dbName = "hivedb";
+ private String tableName = "testTable";
+ private GobblinMCEWriter gobblinMCEWriter;
+ private KafkaStreamingExtractor.KafkaWatermark watermark;
+
+ private GobblinMetadataChangeEvent.Builder gmceBuilder;
+
+ // Not using field injection because they must be different classes
+ private MetadataWriter mockWriter;
+ private MetadataWriter exceptionWriter;
+
+ @Mock
+ private FileSystem fs;
+
+ @Mock
+ private HiveSpec mockHiveSpec;
+
+ @Mock
+ private HiveTable mockTable;
+
+ @AfterMethod
+ public void clean() throws Exception {
+ gobblinMCEWriter.close();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ initMocks();
+ gmceBuilder = GobblinMetadataChangeEvent.newBuilder()
+ .setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:namespace:dataPlatform:hdfs")
+ .setNativeName("testDB/testTable")
+ .build())
+ .setFlowId("testFlow")
+ .setSchemaSource(SchemaSource.EVENT)
+ .setOperationType(OperationType.add_files)
+ .setCluster(ClustersNames.getInstance().getClusterName());
+ watermark = new KafkaStreamingExtractor.KafkaWatermark(
+ new KafkaPartition.Builder().withTopicName("GobblinMetadataChangeEvent_test").withId(1).build(),
+ new LongWatermark(10L));
+
+ State state = new State();
+ String metadataWriters = String.join(",",
+ Arrays.asList(mockWriter.getClass().getName(), exceptionWriter.getClass().getName()));
+ state.setProp("gmce.metadata.writer.classes", metadataWriters);
+
+ Mockito.doNothing().when(mockWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.doThrow(new IOException("Test Exception")).when(exceptionWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+
+ PowerMockito.mockStatic(GobblinConstructorUtils.class);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(mockWriter.getClass().getName()), any(State.class)))
+ .thenReturn(mockWriter);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(exceptionWriter.getClass().getName()), any(State.class)))
+ .thenReturn(exceptionWriter);
+
+ PowerMockito.mockStatic(FileSystem.class);
+ when(FileSystem.get(any()))
+ .thenReturn(fs);
+
+ when(mockTable.getDbName()).thenReturn(dbName);
+ when(mockTable.getTableName()).thenReturn(tableName);
+ when(mockHiveSpec.getTable()).thenReturn(mockTable);
+
+ gobblinMCEWriter = new GobblinMCEWriter(new GobblinMCEWriterBuilder(), state);
+ }
+
+ @Test
+ public void testWriteWhenWriterSpecified() throws IOException {
+ gmceBuilder.setAllowedMetadataWriters(Arrays.asList(mockWriter.getClass().getName()));
+ writeToMetadataWriters(gmceBuilder.build());
+
+ Mockito.verify(mockWriter, Mockito.times(1)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, never()).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ }
+
+ @Test
+ public void testFaultTolerance() throws IOException {
Review Comment:
This test kind of already exists in the Iceberg writer test but I felt the testing could be done more explicitly here without the fluff related to iceberg / hive
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r843197001
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java:
##########
@@ -294,12 +298,14 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
}
// Add fault tolerant ability and make sure we can emit GTE as desired
- private void write(RecordEnvelope recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
+ @VisibleForTesting
+ void write(RecordEnvelope<GenericRecord> recordEnvelope, ConcurrentHashMap newSpecsMap, ConcurrentHashMap oldSpecsMap, HiveSpec spec) throws IOException {
Review Comment:
Entry point for calling writers
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r843197743
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriterTest.java:
##########
@@ -0,0 +1,166 @@
+/*
+ * 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.iceberg.writer;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.hive.HiveTable;
+import org.apache.gobblin.hive.spec.HiveSpec;
+import org.apache.gobblin.hive.writer.MetadataWriter;
+import org.apache.gobblin.metadata.DatasetIdentifier;
+import org.apache.gobblin.metadata.GobblinMetadataChangeEvent;
+import org.apache.gobblin.metadata.OperationType;
+import org.apache.gobblin.metadata.SchemaSource;
+import org.apache.gobblin.source.extractor.extract.LongWatermark;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaPartition;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaStreamingExtractor;
+import org.apache.gobblin.stream.RecordEnvelope;
+import org.apache.gobblin.util.ClustersNames;
+import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.powermock.modules.testng.PowerMockTestCase;
+import org.testng.IObjectFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Test class uses PowerMockito and Testng
+ * References:
+ * https://github.com/powermock/powermock/issues/434
+ * https://jivimberg.io/blog/2016/04/03/using-powermock-plus-testng-to-mock-a-static-class/
+ * https://www.igorkromin.net/index.php/2018/10/04/how-to-fix-powermock-exception-linkageerror-loader-constraint-violation/
+ */
+@PrepareForTest({GobblinConstructorUtils.class, FileSystem.class})
+@PowerMockIgnore("javax.management.*")
+public class GobblinMCEWriterTest extends PowerMockTestCase {
+
+ private String dbName = "hivedb";
+ private String tableName = "testTable";
+ private GobblinMCEWriter gobblinMCEWriter;
+ private KafkaStreamingExtractor.KafkaWatermark watermark;
+
+ private GobblinMetadataChangeEvent.Builder gmceBuilder;
+
+ @Mock
+ private MetadataWriter mockWriter;
+
+ @Mock
+ private FileSystem fs;
+
+ @Mock
+ private HiveSpec mockHiveSpec;
+
+ @AfterMethod
+ public void clean() throws Exception {
+ gobblinMCEWriter.close();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ gmceBuilder = GobblinMetadataChangeEvent.newBuilder()
+ .setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:li:dataPlatform:hdfs")
+ .setNativeName("/data/tracking/testTable")
+ .build())
+ .setFlowId("testFlow")
+ .setSchemaSource(SchemaSource.EVENT)
+ .setOperationType(OperationType.add_files)
+ .setCluster(ClustersNames.getInstance().getClusterName());
+ watermark = new KafkaStreamingExtractor.KafkaWatermark(
+ new KafkaPartition.Builder().withTopicName("GobblinMetadataChangeEvent_test").withId(1).build(),
+ new LongWatermark(10L));
+
+ State state = new State();
+ state.setProp("gmce.metadata.writer.classes", mockWriter.getClass().getName());
+ Mockito.doNothing().when(mockWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+
+ PowerMockito.mockStatic(GobblinConstructorUtils.class);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(mockWriter.getClass().getName()), any(State.class)))
+ .thenReturn(mockWriter);
+
+ PowerMockito.mockStatic(FileSystem.class);
+ when(FileSystem.get(any()))
+ .thenReturn(fs);
+
+ HiveTable mockTable = Mockito.mock(HiveTable.class);
+ when(mockTable.getDbName()).thenReturn(dbName);
+ when(mockTable.getTableName()).thenReturn(tableName);
+
+ mockHiveSpec = Mockito.mock(HiveSpec.class);
+ when(mockHiveSpec.getTable()).thenReturn(mockTable);
+
+ gobblinMCEWriter = new GobblinMCEWriter(new GobblinMCEWriterBuilder(), state);
+ }
+
+ @Test
+ public void testWriteNoOpIfNotSpecified() throws IOException {
+ gmceBuilder.setAllowedMetadataWriters(Arrays.asList("Some other metadata writer"));
+ gobblinMCEWriter.write(
+ new RecordEnvelope<>(gmceBuilder.build(), watermark),
+ new ConcurrentHashMap(), new ConcurrentHashMap(), mockHiveSpec);
+
+ Mockito.verify(mockWriter, never()).writeEnvelope(any(), any(), any(), any());
+ }
+
+ @Test(dataProvider = "AllowMockMetadataWriter")
Review Comment:
We expect calls to the mock writer if the class name is specified in the GMCE or there are no metadata writers specified.
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r844704274
##########
gobblin-iceberg/src/test/resources/publisherTest/data.json:
##########
@@ -1,8 +1,10 @@
-{
- "id": 1,
- "name": "Alyssa"
-}
-{
- "id": 2,
- "name": "Bob"
-}
\ No newline at end of file
+[
+ {
Review Comment:
Cleaning up "invalid" json syntax. I think the reason this worked before is because of Jackson magic
--
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] homatthew commented on a diff in pull request #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
homatthew commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r844706963
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriterTest.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.iceberg.writer;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
+import lombok.SneakyThrows;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.hive.HiveTable;
+import org.apache.gobblin.hive.spec.HiveSpec;
+import org.apache.gobblin.hive.writer.HiveMetadataWriter;
+import org.apache.gobblin.hive.writer.MetadataWriter;
+import org.apache.gobblin.metadata.DatasetIdentifier;
+import org.apache.gobblin.metadata.GobblinMetadataChangeEvent;
+import org.apache.gobblin.metadata.OperationType;
+import org.apache.gobblin.metadata.SchemaSource;
+import org.apache.gobblin.source.extractor.extract.LongWatermark;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaPartition;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaStreamingExtractor;
+import org.apache.gobblin.stream.RecordEnvelope;
+import org.apache.gobblin.util.ClustersNames;
+import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.powermock.modules.testng.PowerMockTestCase;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Test class uses PowerMockito and Testng
+ * References:
+ * https://github.com/powermock/powermock/issues/434
+ * https://jivimberg.io/blog/2016/04/03/using-powermock-plus-testng-to-mock-a-static-class/
+ * https://www.igorkromin.net/index.php/2018/10/04/how-to-fix-powermock-exception-linkageerror-loader-constraint-violation/
+ */
+@PrepareForTest({GobblinConstructorUtils.class, FileSystem.class})
+@PowerMockIgnore("javax.management.*")
+public class GobblinMCEWriterTest extends PowerMockTestCase {
+
+ private String dbName = "hivedb";
+ private String tableName = "testTable";
+ private GobblinMCEWriter gobblinMCEWriter;
+ private KafkaStreamingExtractor.KafkaWatermark watermark;
+
+ private GobblinMetadataChangeEvent.Builder gmceBuilder;
+
+ // Not using field injection because they must be different classes
+ private MetadataWriter mockWriter;
+ private MetadataWriter exceptionWriter;
+
+ @Mock
+ private FileSystem fs;
+
+ @Mock
+ private HiveSpec mockHiveSpec;
+
+ @Mock
+ private HiveTable mockTable;
+
+ @AfterMethod
+ public void clean() throws Exception {
+ gobblinMCEWriter.close();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ initMocks();
+ gmceBuilder = GobblinMetadataChangeEvent.newBuilder()
+ .setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:namespace:dataPlatform:hdfs")
+ .setNativeName("testDB/testTable")
+ .build())
+ .setFlowId("testFlow")
+ .setSchemaSource(SchemaSource.EVENT)
+ .setOperationType(OperationType.add_files)
+ .setCluster(ClustersNames.getInstance().getClusterName());
+ watermark = new KafkaStreamingExtractor.KafkaWatermark(
+ new KafkaPartition.Builder().withTopicName("GobblinMetadataChangeEvent_test").withId(1).build(),
+ new LongWatermark(10L));
+
+ State state = new State();
+ String metadataWriters = String.join(",",
+ Arrays.asList(mockWriter.getClass().getName(), exceptionWriter.getClass().getName()));
+ state.setProp("gmce.metadata.writer.classes", metadataWriters);
+
+ Mockito.doNothing().when(mockWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.doThrow(new IOException("Test Exception")).when(exceptionWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+
+ PowerMockito.mockStatic(GobblinConstructorUtils.class);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(mockWriter.getClass().getName()), any(State.class)))
+ .thenReturn(mockWriter);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(exceptionWriter.getClass().getName()), any(State.class)))
+ .thenReturn(exceptionWriter);
+
+ PowerMockito.mockStatic(FileSystem.class);
+ when(FileSystem.get(any()))
+ .thenReturn(fs);
+
+ when(mockTable.getDbName()).thenReturn(dbName);
+ when(mockTable.getTableName()).thenReturn(tableName);
+ when(mockHiveSpec.getTable()).thenReturn(mockTable);
+
+ gobblinMCEWriter = new GobblinMCEWriter(new GobblinMCEWriterBuilder(), state);
+ }
+
+ @Test
+ public void testWriteWhenWriterSpecified() throws IOException {
+ gmceBuilder.setAllowedMetadataWriters(Arrays.asList(mockWriter.getClass().getName()));
+ writeToMetadataWriters(gmceBuilder.build());
+
+ Mockito.verify(mockWriter, Mockito.times(1)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, never()).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ }
+
+ @Test
+ public void testFaultTolerance() throws IOException {
+
+ gobblinMCEWriter.setMaxErrorDataset(1);
+ gobblinMCEWriter.metadataWriters = Arrays.asList(mockWriter, exceptionWriter, mockWriter);
+ gobblinMCEWriter.tableOperationTypeMap = new HashMap<>();
+
+ String dbName2 = dbName + "2";
+ String otherDb = "someOtherDB";
+ addTableStatus(dbName, "datasetPath");
+ addTableStatus(dbName2, "datasetPath");
+ addTableStatus(otherDb, "otherDatasetPath");
+
+ BiFunction<String, String, Void> verifyMocksCalled = new BiFunction<String, String, Void>(){
+ private int timesCalled = 0;
+ @Override
+ @SneakyThrows
+ public Void apply(String dbName, String tableName) {
+ timesCalled++;
+
+ // also validates that order is maintained since all writers after an exception should reset instead of write
+ Mockito.verify(mockWriter, Mockito.times(timesCalled)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, Mockito.times(timesCalled)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, Mockito.times(1)).reset(dbName, tableName);
+ Mockito.verify(mockWriter, Mockito.times(1)).reset(dbName, tableName);
+ return null;
+ }
+ };
+
+ writeToMetadataWriters(gmceBuilder.build());
+ verifyMocksCalled.apply(dbName, tableName);
+
+ // Another exception for same dataset but different db
+ when(mockTable.getDbName()).thenReturn(dbName2);
+ writeToMetadataWriters(gmceBuilder.build());
+ verifyMocksCalled.apply(dbName2, tableName);
+
+ // exception thrown because exceeds max number of datasets with errors
+ when(mockTable.getDbName()).thenReturn(otherDb);
+ Assert.expectThrows(IOException.class, () -> writeToMetadataWriters(gmceBuilder.setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:namespace:dataPlatform:hdfs")
+ .setNativeName("someOtherDB/testTable")
+ .build()).build()));
+ }
+
+ @Test(dataProvider = "AllowMockMetadataWriter")
+ public void testGetAllowedMetadataWriters(List<String> metadataWriters) {
+ Assert.assertNotEquals(mockWriter.getClass().getName(), exceptionWriter.getClass().getName());
+ gmceBuilder.setAllowedMetadataWriters(metadataWriters);
+ List<MetadataWriter> allowedWriters = GobblinMCEWriter.getAllowedMetadataWriters(
+ gmceBuilder.build(),
+ Arrays.asList(mockWriter, exceptionWriter));
+
+ Assert.assertEquals(allowedWriters.size(), 2);
+ Assert.assertEquals(allowedWriters.get(0).getClass().getName(), mockWriter.getClass().getName());
+ Assert.assertEquals(allowedWriters.get(1).getClass().getName(), exceptionWriter.getClass().getName());
+ }
+
+ @DataProvider(name="AllowMockMetadataWriter")
+ public Object[][] allowMockMetadataWriterParams() {
+ initMocks();
+ return new Object[][] {
+ {Arrays.asList(mockWriter.getClass().getName(), exceptionWriter.getClass().getName())},
+ {Collections.emptyList()}
+ };
+ }
+
+ private void initMocks() {
+ MockitoAnnotations.initMocks(this);
+ mockWriter = Mockito.mock(MetadataWriter.class);
+ exceptionWriter = Mockito.mock(HiveMetadataWriter.class);
Review Comment:
This is kind of a hack so I could have 2 `MetadataWriter` mocks with different class names
##########
gobblin-iceberg/src/test/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriterTest.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * 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.iceberg.writer;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
+import lombok.SneakyThrows;
+import org.apache.gobblin.configuration.State;
+import org.apache.gobblin.hive.HiveTable;
+import org.apache.gobblin.hive.spec.HiveSpec;
+import org.apache.gobblin.hive.writer.HiveMetadataWriter;
+import org.apache.gobblin.hive.writer.MetadataWriter;
+import org.apache.gobblin.metadata.DatasetIdentifier;
+import org.apache.gobblin.metadata.GobblinMetadataChangeEvent;
+import org.apache.gobblin.metadata.OperationType;
+import org.apache.gobblin.metadata.SchemaSource;
+import org.apache.gobblin.source.extractor.extract.LongWatermark;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaPartition;
+import org.apache.gobblin.source.extractor.extract.kafka.KafkaStreamingExtractor;
+import org.apache.gobblin.stream.RecordEnvelope;
+import org.apache.gobblin.util.ClustersNames;
+import org.apache.gobblin.util.reflection.GobblinConstructorUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.testng.PowerMockObjectFactory;
+import org.powermock.modules.testng.PowerMockTestCase;
+import org.testng.Assert;
+import org.testng.IObjectFactory;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.ObjectFactory;
+import org.testng.annotations.Test;
+
+import static org.mockito.Mockito.*;
+
+
+/**
+ * Test class uses PowerMockito and Testng
+ * References:
+ * https://github.com/powermock/powermock/issues/434
+ * https://jivimberg.io/blog/2016/04/03/using-powermock-plus-testng-to-mock-a-static-class/
+ * https://www.igorkromin.net/index.php/2018/10/04/how-to-fix-powermock-exception-linkageerror-loader-constraint-violation/
+ */
+@PrepareForTest({GobblinConstructorUtils.class, FileSystem.class})
+@PowerMockIgnore("javax.management.*")
+public class GobblinMCEWriterTest extends PowerMockTestCase {
+
+ private String dbName = "hivedb";
+ private String tableName = "testTable";
+ private GobblinMCEWriter gobblinMCEWriter;
+ private KafkaStreamingExtractor.KafkaWatermark watermark;
+
+ private GobblinMetadataChangeEvent.Builder gmceBuilder;
+
+ // Not using field injection because they must be different classes
+ private MetadataWriter mockWriter;
+ private MetadataWriter exceptionWriter;
+
+ @Mock
+ private FileSystem fs;
+
+ @Mock
+ private HiveSpec mockHiveSpec;
+
+ @Mock
+ private HiveTable mockTable;
+
+ @AfterMethod
+ public void clean() throws Exception {
+ gobblinMCEWriter.close();
+ }
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ initMocks();
+ gmceBuilder = GobblinMetadataChangeEvent.newBuilder()
+ .setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:namespace:dataPlatform:hdfs")
+ .setNativeName("testDB/testTable")
+ .build())
+ .setFlowId("testFlow")
+ .setSchemaSource(SchemaSource.EVENT)
+ .setOperationType(OperationType.add_files)
+ .setCluster(ClustersNames.getInstance().getClusterName());
+ watermark = new KafkaStreamingExtractor.KafkaWatermark(
+ new KafkaPartition.Builder().withTopicName("GobblinMetadataChangeEvent_test").withId(1).build(),
+ new LongWatermark(10L));
+
+ State state = new State();
+ String metadataWriters = String.join(",",
+ Arrays.asList(mockWriter.getClass().getName(), exceptionWriter.getClass().getName()));
+ state.setProp("gmce.metadata.writer.classes", metadataWriters);
+
+ Mockito.doNothing().when(mockWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.doThrow(new IOException("Test Exception")).when(exceptionWriter)
+ .writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+
+ PowerMockito.mockStatic(GobblinConstructorUtils.class);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(mockWriter.getClass().getName()), any(State.class)))
+ .thenReturn(mockWriter);
+ when(GobblinConstructorUtils.invokeConstructor(
+ eq(MetadataWriter.class), eq(exceptionWriter.getClass().getName()), any(State.class)))
+ .thenReturn(exceptionWriter);
+
+ PowerMockito.mockStatic(FileSystem.class);
+ when(FileSystem.get(any()))
+ .thenReturn(fs);
+
+ when(mockTable.getDbName()).thenReturn(dbName);
+ when(mockTable.getTableName()).thenReturn(tableName);
+ when(mockHiveSpec.getTable()).thenReturn(mockTable);
+
+ gobblinMCEWriter = new GobblinMCEWriter(new GobblinMCEWriterBuilder(), state);
+ }
+
+ @Test
+ public void testWriteWhenWriterSpecified() throws IOException {
+ gmceBuilder.setAllowedMetadataWriters(Arrays.asList(mockWriter.getClass().getName()));
+ writeToMetadataWriters(gmceBuilder.build());
+
+ Mockito.verify(mockWriter, Mockito.times(1)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, never()).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ }
+
+ @Test
+ public void testFaultTolerance() throws IOException {
+
+ gobblinMCEWriter.setMaxErrorDataset(1);
+ gobblinMCEWriter.metadataWriters = Arrays.asList(mockWriter, exceptionWriter, mockWriter);
+ gobblinMCEWriter.tableOperationTypeMap = new HashMap<>();
+
+ String dbName2 = dbName + "2";
+ String otherDb = "someOtherDB";
+ addTableStatus(dbName, "datasetPath");
+ addTableStatus(dbName2, "datasetPath");
+ addTableStatus(otherDb, "otherDatasetPath");
+
+ BiFunction<String, String, Void> verifyMocksCalled = new BiFunction<String, String, Void>(){
+ private int timesCalled = 0;
+ @Override
+ @SneakyThrows
+ public Void apply(String dbName, String tableName) {
+ timesCalled++;
+
+ // also validates that order is maintained since all writers after an exception should reset instead of write
+ Mockito.verify(mockWriter, Mockito.times(timesCalled)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, Mockito.times(timesCalled)).writeEnvelope(
+ Mockito.any(RecordEnvelope.class), Mockito.anyMap(), Mockito.anyMap(), Mockito.any(HiveSpec.class));
+ Mockito.verify(exceptionWriter, Mockito.times(1)).reset(dbName, tableName);
+ Mockito.verify(mockWriter, Mockito.times(1)).reset(dbName, tableName);
+ return null;
+ }
+ };
+
+ writeToMetadataWriters(gmceBuilder.build());
+ verifyMocksCalled.apply(dbName, tableName);
+
+ // Another exception for same dataset but different db
+ when(mockTable.getDbName()).thenReturn(dbName2);
+ writeToMetadataWriters(gmceBuilder.build());
+ verifyMocksCalled.apply(dbName2, tableName);
+
+ // exception thrown because exceeds max number of datasets with errors
+ when(mockTable.getDbName()).thenReturn(otherDb);
+ Assert.expectThrows(IOException.class, () -> writeToMetadataWriters(gmceBuilder.setDatasetIdentifier(DatasetIdentifier.newBuilder()
+ .setDataPlatformUrn("urn:namespace:dataPlatform:hdfs")
+ .setNativeName("someOtherDB/testTable")
+ .build()).build()));
+ }
+
+ @Test(dataProvider = "AllowMockMetadataWriter")
+ public void testGetAllowedMetadataWriters(List<String> metadataWriters) {
+ Assert.assertNotEquals(mockWriter.getClass().getName(), exceptionWriter.getClass().getName());
+ gmceBuilder.setAllowedMetadataWriters(metadataWriters);
+ List<MetadataWriter> allowedWriters = GobblinMCEWriter.getAllowedMetadataWriters(
+ gmceBuilder.build(),
+ Arrays.asList(mockWriter, exceptionWriter));
+
+ Assert.assertEquals(allowedWriters.size(), 2);
+ Assert.assertEquals(allowedWriters.get(0).getClass().getName(), mockWriter.getClass().getName());
+ Assert.assertEquals(allowedWriters.get(1).getClass().getName(), exceptionWriter.getClass().getName());
+ }
+
+ @DataProvider(name="AllowMockMetadataWriter")
+ public Object[][] allowMockMetadataWriterParams() {
+ initMocks();
+ return new Object[][] {
+ {Arrays.asList(mockWriter.getClass().getName(), exceptionWriter.getClass().getName())},
+ {Collections.emptyList()}
+ };
+ }
+
+ private void initMocks() {
+ MockitoAnnotations.initMocks(this);
+ mockWriter = Mockito.mock(MetadataWriter.class);
+ exceptionWriter = Mockito.mock(HiveMetadataWriter.class);
Review Comment:
This is a hack so I could have 2 `MetadataWriter` mocks with different class names
--
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 #3490: [GOBBLIN-1613] Add metadata writers field to GMCE schema
Posted by GitBox <gi...@apache.org>.
ZihanLi58 commented on code in PR #3490:
URL: https://github.com/apache/gobblin/pull/3490#discussion_r847628476
##########
gobblin-iceberg/src/main/java/org/apache/gobblin/iceberg/writer/GobblinMCEWriter.java:
##########
@@ -288,18 +292,37 @@ public void writeEnvelope(RecordEnvelope<GenericRecord> recordEnvelope) throws I
((LongWatermark)watermark.getWatermark()).getValue()-1, ((LongWatermark)watermark.getWatermark()).getValue()));
}
tableOperationTypeMap.get(tableString).gmceHighWatermark = ((LongWatermark)watermark.getWatermark()).getValue();
- write(recordEnvelope, newSpecsMap, oldSpecsMap, spec);
+
+ List<MetadataWriter> allowedWriters = getAllowedMetadataWriters(gmce, metadataWriters);
+ writeToMetadataWriters(recordEnvelope, allowedWriters, newSpecsMap, oldSpecsMap, spec);
Review Comment:
writeWithMetadataWriters?
--
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