You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@seatunnel.apache.org by GitBox <gi...@apache.org> on 2022/06/20 06:53:30 UTC
[GitHub] [incubator-seatunnel] lhyundeadsoul opened a new pull request, #2036: [Feature][Connector] add IT for Assert Sink in e2e module
lhyundeadsoul opened a new pull request, #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036
<!--
Thank you for contributing to SeaTunnel! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
## Contribution Checklist
- Make sure that the pull request corresponds to a [GITHUB issue](https://github.com/apache/incubator-seatunnel/issues).
- Name the pull request in the form "[Feature] [component] Title of the pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Minor fixes should be named following this pattern: `[hotfix] [docs] Fix typo in README.md doc`.
-->
## Purpose of this pull request
add IT to test Assert Sink plugin
https://github.com/apache/incubator-seatunnel/issues/2034
<!-- Describe the purpose of this pull request. For example: This pull request adds checkstyle plugin.-->
## Check list
* [x] Code changed are covered with tests, or it does not need tests for reason:
* [x] If any new Jar binary package adding in your PR, please add License Notice according
[New License Guide](https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/contribution/new-license.md)
* [x] If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
Hisoka-X commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171044283
> Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
>
> ```
> 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
>
>
> java.lang.AssertionError:
> Expected :0
> Actual :1
> <Click to see difference>
> ```
So this error have any progress?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
Hisoka-X commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1170980348
> Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
>
> ```
> 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
>
>
> java.lang.AssertionError:
> Expected :0
> Actual :1
> <Click to see difference>
> ```
In your local?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910908653
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
We can't use `getBatchEnvironment().getExecutionPlan()` after `collect()`
Using collect instead of print will cause:
```
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Execute Flink task error
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.RuntimeException: Execute Flink task error
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:84)
at org.apache.seatunnel.core.base.Seatunnel.run(Seatunnel.java:39)
at org.apache.seatunnel.core.flink.SeatunnelFlink.main(SeatunnelFlink.java:33)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 8 more
Caused by: java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1165)
at org.apache.flink.api.java.ExecutionEnvironment.getExecutionPlan(ExecutionEnvironment.java:1076)
at org.apache.seatunnel.flink.batch.FlinkBatchExecution.start(FlinkBatchExecution.java:76)
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:81)
... 15 more
```
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171041531
> > > > > > Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
> > > > > > ```
> > > > > > 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
> > > > > >
> > > > > >
> > > > > > java.lang.AssertionError:
> > > > > > Expected :0
> > > > > > Actual :1
> > > > > > <Click to see difference>
> > > > > > ```
> > > > >
> > > > >
> > > > > In your local?
> > > > > Maybe it's in my local.
> > > >
> > > >
> > > > I find the root cause : There is no `org.apache.seatunnel.core.flink.FlinkEnvParameterParser` in my target jar. and an error often occur while execute 'mvn install'
> > > > > Failure to find org.apache.seatunnel:seatunnel:pom:${revision} in https://repo.maven.apache.org/maven2 was cached in the local repository, resolution will not be reattempted until the update interval of central has elapsed or updates are forced
> > > >
> > > >
> > > > Have you ever had this?
> > >
> > >
> > > Yes, also confusion meðŸ˜
> >
> >
> > Wow~ I am not the only one~ HELP! @ruanwenjun Have you ever had this error?
>
> I doubt it is caused by `flatten-maven-plugin`. But I don't have test.
Maybe. Sometimes it work, sometimes it fail. Occur randomly.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910930398
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
`ConsoleSink` can pass CI because it was exclusive by `org.apache.seatunnel.flink.batch.FlinkBatchExecution#whetherExecute`
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171012313
> > > > Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
> > > > ```
> > > > 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
> > > >
> > > >
> > > > java.lang.AssertionError:
> > > > Expected :0
> > > > Actual :1
> > > > <Click to see difference>
> > > > ```
> > >
> > >
> > > In your local?
> > > Maybe it's in my local.
> >
> >
> > I find the root cause : There is no `org.apache.seatunnel.core.flink.FlinkEnvParameterParser` in my target jar. and an error often occur while execute 'mvn install'
> > > Failure to find org.apache.seatunnel:seatunnel:pom:${revision} in https://repo.maven.apache.org/maven2 was cached in the local repository, resolution will not be reattempted until the update interval of central has elapsed or updates are forced
> >
> >
> > Have you ever had this?
>
> Yes, also confusion meðŸ˜
Wow~ I am not the only one~
HELP! @ruanwenjun Have you ever had this error?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910908653
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
We can't use `getBatchEnvironment().getExecutionPlan()` after `collect()`
Using collect instead of print will cause:
```
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Execute Flink task error
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.RuntimeException: Execute Flink task error
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:84)
at org.apache.seatunnel.core.base.Seatunnel.run(Seatunnel.java:39)
at org.apache.seatunnel.core.flink.SeatunnelFlink.main(SeatunnelFlink.java:33)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 8 more
Caused by: java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1165)
at org.apache.flink.api.java.ExecutionEnvironment.getExecutionPlan(ExecutionEnvironment.java:1076)
at org.apache.seatunnel.flink.batch.FlinkBatchExecution.start(FlinkBatchExecution.java:76)
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:81)
... 15 more
```
If `ConsoleSink` doesn't catch Exception when using `print()` , it will also fail.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171161804
@Hisoka-X PTAL
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r904938061
##########
seatunnel-e2e/seatunnel-flink-e2e/src/test/resources/assertion/fakesource_to_assert.conf:
##########
@@ -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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ field_name = "name,age"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+ sql {
+ sql = "select name,age from fake"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+ AssertSink {
+ rules =
+ [{
+ field_name = name
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN_LENGTH
+ rule_value = 3
+ },
+ {
+ rule_type = MAX_LENGTH
+ rule_value = 5
+ }
+ ]
+ },{
+ field_name = age
+ field_type = int
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN
+ rule_value = 10
+ },
+ {
+ rule_type = MAX
+ rule_value = 20
+ }
+ ]
+ }
Review Comment:
You may need to modify this length, I see the default mock name of FakeSource is
```
"Gary", "Ricky Huo", "Kid Xiong"
```
and the age is between 1-100.
Or you need to set a mock schema.
https://seatunnel.apache.org/docs/connector/source/Fake#mock_data_schema-option--listcolumn_config
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1164133988
> @lhyundeadsoul Is there any count rule? e.g. I want to assert the batch row count is 4.
Thanks for your idea.
I support five assert rule now: min/max/min_length/max_length/not null , and will consider count rule in another PR.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910908653
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
Using collect instead of print will cause:
```
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Execute Flink task error
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.RuntimeException: Execute Flink task error
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:84)
at org.apache.seatunnel.core.base.Seatunnel.run(Seatunnel.java:39)
at org.apache.seatunnel.core.flink.SeatunnelFlink.main(SeatunnelFlink.java:33)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 8 more
Caused by: java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1165)
at org.apache.flink.api.java.ExecutionEnvironment.getExecutionPlan(ExecutionEnvironment.java:1076)
at org.apache.seatunnel.flink.batch.FlinkBatchExecution.start(FlinkBatchExecution.java:76)
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:81)
... 15 more
```
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910957390
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
OK, I got you, since when we execute collect, the flink job will be executed, so we cannot execute it again.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1164274555
The IT failed.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171045304
>
a new error occur
```
Caused by: java.lang.RuntimeException: No data sinks have been created yet. A program needs at least one sink that consumes data. Examples are writing the data set or printing it.
at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1170)
at org.apache.flink.api.java.ExecutionEnvironment.getExecutionPlan(ExecutionEnvironment.java:1076)
at org.apache.seatunnel.flink.batch.FlinkBatchExecution.start(FlinkBatchExecution.java:76)
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:81)
... 15 more
```
I working on it
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1170976548
/Users/lihongyuinfo/Library/Java/JavaVirtualMachines/openjdk-15.0.1/Contents/Home/bin/java -ea -Djacoco-agent.destfile=/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-e2e/seatunnel-flink-e2e/target/jacoco.exec -Didea.test.cyclic.buffer.size=1048576 -javaagent:/Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=56571:/Applications/IntelliJ IDEA.app/Contents/bin -Dfile.encoding=UTF-8 -classpath /Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar:/Applications/IntelliJ IDEA.app/Contents/plugins/junit/lib/junit5-rt.jar:/Applications/IntelliJ IDEA.app/Contents/plugins/junit/lib/junit-rt.jar:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-e2e/seatunnel-flink-e2e/target/test-classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-e2e/seatunnel-flink-e2e/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-core/seatunnel-core-flink/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-core/seatunnel-core-base/t
arget/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-apis/seatunnel-api-base/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-apis/seatunnel-api-spark/target/classes:/Users/lihongyuinfo/.m2/repository/com/beust/jcommander/1.81/jcommander-1.81.jar:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-apis/seatunnel-api-flink/target/classes:/Users/lihongyuinfo/.m2/repository/org/apache/flink/flink-statebackend-rocksdb_2.11/1.13.6/flink-statebackend-rocksdb_2.11-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/com/ververica/frocksdbjni/5.17.2-ververica-2.1/frocksdbjni-5.17.2-ververica-2.1.jar:/Users/lihongyuinfo/.m2/repository/com/google/code/findbugs/jsr305/1.3.9/jsr305-1.3.9.jar:/Users/lihongyuinfo/.m2/repository/org/apache/flink/force-shading/1.13.6/force-shading-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/flink/flink-csv/1.13.6/flink-csv-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/flink/flink-orc_2.11/1.13.6/flin
k-orc_2.11-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/orc/orc-core/1.5.6/orc-core-1.5.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/orc/orc-shims/1.5.6/orc-shims-1.5.6.jar:/Users/lihongyuinfo/.m2/repository/com/google/protobuf/protobuf-java/2.5.0/protobuf-java-2.5.0.jar:/Users/lihongyuinfo/.m2/repository/commons-lang/commons-lang/2.6/commons-lang-2.6.jar:/Users/lihongyuinfo/.m2/repository/io/airlift/aircompressor/0.10/aircompressor-0.10.jar:/Users/lihongyuinfo/.m2/repository/org/apache/hive/hive-storage-api/2.6.0/hive-storage-api-2.6.0.jar:/Users/lihongyuinfo/.m2/repository/org/apache/flink/flink-parquet_2.11/1.13.6/flink-parquet_2.11-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/parquet/parquet-hadoop/1.11.1/parquet-hadoop-1.11.1.jar:/Users/lihongyuinfo/.m2/repository/org/apache/parquet/parquet-column/1.11.1/parquet-column-1.11.1.jar:/Users/lihongyuinfo/.m2/repository/org/apache/parquet/parquet-common/1.11.1/parquet-common-1.11.1.jar:/Users/lihongyuin
fo/.m2/repository/org/apache/yetus/audience-annotations/0.11.0/audience-annotations-0.11.0.jar:/Users/lihongyuinfo/.m2/repository/org/apache/parquet/parquet-encoding/1.11.1/parquet-encoding-1.11.1.jar:/Users/lihongyuinfo/.m2/repository/org/apache/parquet/parquet-format-structures/1.11.1/parquet-format-structures-1.11.1.jar:/Users/lihongyuinfo/.m2/repository/javax/annotation/javax.annotation-api/1.3.2/javax.annotation-api-1.3.2.jar:/Users/lihongyuinfo/.m2/repository/org/apache/parquet/parquet-jackson/1.11.1/parquet-jackson-1.11.1.jar:/Users/lihongyuinfo/.m2/repository/commons-pool/commons-pool/1.6/commons-pool-1.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/flink/flink-json/1.13.6/flink-json-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/flink/flink-avro/1.13.6/flink-avro-1.13.6.jar:/Users/lihongyuinfo/.m2/repository/org/apache/avro/avro/1.10.0/avro-1.10.0.jar:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-transforms/seatunnel-transforms-flink/seatunnel-tra
nsform-flink-sql/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-transforms/seatunnel-transforms-flink/seatunnel-transform-flink-split/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-transforms/seatunnel-transforms-flink/seatunnel-transform-flink-table2datastream/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-transforms/seatunnel-transforms-flink/seatunnel-transform-flink-datastream2table/target/classes:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-transforms/seatunnel-transforms-flink/seatunnel-transform-flink-udf/target/classes:/Users/lihongyuinfo/.m2/repository/org/slf4j/slf4j-log4j12/1.7.25/slf4j-log4j12-1.7.25.jar:/Users/lihongyuinfo/.m2/repository/org/slf4j/slf4j-api/1.7.25/slf4j-api-1.7.25.jar:/Users/lihongyuinfo/.m2/repository/log4j/log4j/1.2.17/log4j-1.2.17.jar:/Users/lihongyuinfo/.m2/repository/org/testcontainers/testcontainers/1.16.3/testcontainers-1.16.3.jar:/Users/lihongyuinfo/.m2/re
pository/org/apache/commons/commons-compress/1.21/commons-compress-1.21.jar:/Users/lihongyuinfo/.m2/repository/org/rnorth/duct-tape/duct-tape/1.0.8/duct-tape-1.0.8.jar:/Users/lihongyuinfo/.m2/repository/org/jetbrains/annotations/17.0.0/annotations-17.0.0.jar:/Users/lihongyuinfo/.m2/repository/com/github/docker-java/docker-java-api/3.2.12/docker-java-api-3.2.12.jar:/Users/lihongyuinfo/.m2/repository/com/fasterxml/jackson/core/jackson-annotations/2.12.6/jackson-annotations-2.12.6.jar:/Users/lihongyuinfo/.m2/repository/com/github/docker-java/docker-java-transport-zerodep/3.2.12/docker-java-transport-zerodep-3.2.12.jar:/Users/lihongyuinfo/.m2/repository/com/github/docker-java/docker-java-transport/3.2.12/docker-java-transport-3.2.12.jar:/Users/lihongyuinfo/.m2/repository/net/java/dev/jna/jna/5.8.0/jna-5.8.0.jar:/Users/lihongyuinfo/Sources/incubator-seatunnel/seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-clickhouse/target/classes:/Users/lihongyuinfo/Sources/in
cubator-seatunnel/seatunnel-common/target/classes:/Users/lihongyuinfo/.m2/repository/org/apache/seatunnel/seatunnel-config-shade/2.1.1/seatunnel-config-shade-2.1.1.jar:/Users/lihongyuinfo/.m2/repository/org/apache/seatunnel/seatunnel-config-base/2.1.1/seatunnel-config-base-2.1.1.jar:/Users/lihongyuinfo/.m2/repository/com/typesafe/config/1.3.3/config-1.3.3.jar:/Users/lihongyuinfo/.m2/repository/org/scala-lang/scala-library/2.11.12/scala-library-2.11.12.jar:/Users/lihongyuinfo/.m2/repository/org/apache/commons/commons-lang3/3.4/commons-lang3-3.4.jar:/Users/lihongyuinfo/.m2/repository/com/fasterxml/jackson/core/jackson-databind/2.12.6/jackson-databind-2.12.6.jar:/Users/lihongyuinfo/.m2/repository/ru/yandex/clickhouse/clickhouse-jdbc/0.2/clickhouse-jdbc-0.2.jar:/Users/lihongyuinfo/.m2/repository/org/apache/httpcomponents/httpclient/4.5.13/httpclient-4.5.13.jar:/Users/lihongyuinfo/.m2/repository/org/apache/httpcomponents/httpcore/4.4.4/httpcore-4.4.4.jar:/Users/lihongyuinfo/.m2/repositor
y/commons-logging/commons-logging/1.2/commons-logging-1.2.jar:/Users/lihongyuinfo/.m2/repository/commons-codec/commons-codec/1.11/commons-codec-1.11.jar:/Users/lihongyuinfo/.m2/repository/org/apache/httpcomponents/httpmime/4.5.2/httpmime-4.5.2.jar:/Users/lihongyuinfo/.m2/repository/net/jpountz/lz4/lz4/1.3.0/lz4-1.3.0.jar:/Users/lihongyuinfo/.m2/repository/com/fasterxml/jackson/core/jackson-core/2.12.6/jackson-core-2.12.6.jar:/Users/lihongyuinfo/.m2/repository/javax/xml/bind/jaxb-api/2.3.0/jaxb-api-2.3.0.jar:/Users/lihongyuinfo/.m2/repository/org/apache/sshd/sshd-scp/2.7.0/sshd-scp-2.7.0.jar:/Users/lihongyuinfo/.m2/repository/org/apache/sshd/sshd-core/2.7.0/sshd-core-2.7.0.jar:/Users/lihongyuinfo/.m2/repository/org/apache/sshd/sshd-common/2.7.0/sshd-common-2.7.0.jar:/Users/lihongyuinfo/.m2/repository/org/slf4j/jcl-over-slf4j/1.7.30/jcl-over-slf4j-1.7.30.jar:/Users/lihongyuinfo/.m2/repository/com/google/auto/service/auto-service/1.0.1/auto-service-1.0.1.jar:/Users/lihongyuinfo/.m2/rep
ository/com/google/auto/service/auto-service-annotations/1.0.1/auto-service-annotations-1.0.1.jar:/Users/lihongyuinfo/.m2/repository/com/google/auto/auto-common/1.2/auto-common-1.2.jar:/Users/lihongyuinfo/.m2/repository/com/google/guava/guava/19.0/guava-19.0.jar:/Users/lihongyuinfo/.m2/repository/junit/junit/4.13.2/junit-4.13.2.jar:/Users/lihongyuinfo/.m2/repository/org/hamcrest/hamcrest-core/1.3/hamcrest-core-1.3.jar:/Users/lihongyuinfo/.m2/repository/org/projectlombok/lombok/1.18.0/lombok-1.18.0.jar com.intellij.rt.junit.JUnitStarter -ideVersion5 -junit4 org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT
22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
java.lang.AssertionError:
Expected :0
Actual :1
<Click to see difference>
at org.junit.Assert.fail(Assert.java:89)
at org.junit.Assert.failNotEquals(Assert.java:835)
at org.junit.Assert.assertEquals(Assert.java:647)
at org.junit.Assert.assertEquals(Assert.java:633)
at org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT.testFakeSourceToConsoleSink(FakeSourceToConsoleIT.java:33)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:64)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:564)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:33)
at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171003178
> > Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
> > ```
> > 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
> >
> >
> > java.lang.AssertionError:
> > Expected :0
> > Actual :1
> > <Click to see difference>
> > ```
>
> In your local?
Maybe it's in my local.
I find the root cause : There is no `org.apache.seatunnel.core.flink.FlinkEnvParameterParser` in my target jar. and an error often occur while execute 'mvn install'
>Failure to find org.apache.seatunnel:seatunnel:pom:${revision} in https://repo.maven.apache.org/maven2 was cached in the local repository, resolution will not be reattempted until the update interval of central has elapsed or updates are forced
Have you ever had this?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
Hisoka-X commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171017406
> > > > > Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
> > > > > ```
> > > > > 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
> > > > >
> > > > >
> > > > > java.lang.AssertionError:
> > > > > Expected :0
> > > > > Actual :1
> > > > > <Click to see difference>
> > > > > ```
> > > >
> > > >
> > > > In your local?
> > > > Maybe it's in my local.
> > >
> > >
> > > I find the root cause : There is no `org.apache.seatunnel.core.flink.FlinkEnvParameterParser` in my target jar. and an error often occur while execute 'mvn install'
> > > > Failure to find org.apache.seatunnel:seatunnel:pom:${revision} in https://repo.maven.apache.org/maven2 was cached in the local repository, resolution will not be reattempted until the update interval of central has elapsed or updates are forced
> > >
> > >
> > > Have you ever had this?
> >
> >
> > Yes, also confusion meðŸ˜
>
> Wow~ I am not the only one~ HELP! @ruanwenjun Have you ever had this error?
I doubt it is caused by `flatten-maven-plugin`. But I don't have test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1164134584
> @lhyundeadsoul Is there any count rule? e.g. I want to assert the batch row count is 4.
Thanks for your suggestion.
I support five assert rule now: min/max/min_length/max_length/not null , and will consider count rule in another PR.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r904938061
##########
seatunnel-e2e/seatunnel-flink-e2e/src/test/resources/assertion/fakesource_to_assert.conf:
##########
@@ -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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ field_name = "name,age"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+ sql {
+ sql = "select name,age from fake"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+ AssertSink {
+ rules =
+ [{
+ field_name = name
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN_LENGTH
+ rule_value = 3
+ },
+ {
+ rule_type = MAX_LENGTH
+ rule_value = 5
+ }
+ ]
+ },{
+ field_name = age
+ field_type = int
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN
+ rule_value = 10
+ },
+ {
+ rule_type = MAX
+ rule_value = 20
+ }
+ ]
+ }
Review Comment:
You may need to modify this length, I see the default mock name of FakeSource is
```
"Gary", "Ricky Huo", "Kid Xiong"
```
and the age is between 1-100.
Or you need to set a mock schema.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r905665547
##########
seatunnel-e2e/seatunnel-flink-e2e/src/test/resources/assertion/fakesource_to_assert.conf:
##########
@@ -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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ field_name = "name,age"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+ sql {
+ sql = "select name,age from fake"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+ AssertSink {
+ rules =
+ [{
+ field_name = name
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN_LENGTH
+ rule_value = 3
+ },
+ {
+ rule_type = MAX_LENGTH
+ rule_value = 5
+ }
+ ]
+ },{
+ field_name = age
+ field_type = int
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN
+ rule_value = 10
+ },
+ {
+ rule_type = MAX
+ rule_value = 20
+ }
+ ]
+ }
Review Comment:
Fine. I will amend the assert config first.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r905165957
##########
seatunnel-e2e/seatunnel-flink-e2e/src/test/resources/assertion/fakesource_to_assert.conf:
##########
@@ -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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ field_name = "name,age"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+ sql {
+ sql = "select name,age from fake"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+ AssertSink {
+ rules =
+ [{
+ field_name = name
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN_LENGTH
+ rule_value = 3
+ },
+ {
+ rule_type = MAX_LENGTH
+ rule_value = 5
+ }
+ ]
+ },{
+ field_name = age
+ field_type = int
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN
+ rule_value = 10
+ },
+ {
+ rule_type = MAX
+ rule_value = 20
+ }
+ ]
+ }
Review Comment:
The error is in container, and it doesn't display, this is strange, but you can test on your environment.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1168538749
Could you please make the ci pass.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910908653
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
This will cause:
```
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Execute Flink task error
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.RuntimeException: Execute Flink task error
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:84)
at org.apache.seatunnel.core.base.Seatunnel.run(Seatunnel.java:39)
at org.apache.seatunnel.core.flink.SeatunnelFlink.main(SeatunnelFlink.java:33)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 8 more
Caused by: java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1165)
at org.apache.flink.api.java.ExecutionEnvironment.getExecutionPlan(ExecutionEnvironment.java:1076)
at org.apache.seatunnel.flink.batch.FlinkBatchExecution.start(FlinkBatchExecution.java:76)
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:81)
... 15 more
```
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
Hisoka-X commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171043135
> Maybe. Sometimes it work, sometimes it fail. Occur randomly.
if you package success local, everything will be fine.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910887980
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
It's better to use collect here, if you don't want to print each row.
```
try {
inDataSet.collect().forEach(row -> {
ASSERT_EXECUTOR
.fail(row, assertFieldRules)
.ifPresent(failRule -> {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
});
} catch (IllegalArgumentException illegalArgumentException) {
throw illegalArgumentException;
} catch (Exception ex) {
throw new RuntimeException("AssertSink execute failed", ex);
}
```
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -71,7 +73,7 @@ public void outputStream(FlinkEnvironment env, DataStream<Row> dataStream) {
throw new IllegalStateException("row :" + row + "field name of the fail rule: " + failRule.getFieldName());
});
return null;
- });
+ }).print();
Review Comment:
You don't need to add print() in stream mode.
```suggestion
});
```
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1170977058
`
22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
java.lang.AssertionError:
Expected :0
Actual :1
<Click to see difference>
`
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910908653
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
We can't use `getBatchEnvironment().getExecutionPlan()` after `collect()`
Using collect instead of print will cause:
```
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Execute Flink task error
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:372)
at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.RuntimeException: Execute Flink task error
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:84)
at org.apache.seatunnel.core.base.Seatunnel.run(Seatunnel.java:39)
at org.apache.seatunnel.core.flink.SeatunnelFlink.main(SeatunnelFlink.java:33)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
... 8 more
Caused by: java.lang.RuntimeException: No new data sinks have been defined since the last execution. The last execution refers to the latest call to 'execute()', 'count()', 'collect()', or 'print()'.
at org.apache.flink.api.java.ExecutionEnvironment.createProgramPlan(ExecutionEnvironment.java:1165)
at org.apache.flink.api.java.ExecutionEnvironment.getExecutionPlan(ExecutionEnvironment.java:1076)
at org.apache.seatunnel.flink.batch.FlinkBatchExecution.start(FlinkBatchExecution.java:76)
at org.apache.seatunnel.core.flink.command.FlinkTaskExecuteCommand.execute(FlinkTaskExecuteCommand.java:81)
... 15 more
```
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r905157662
##########
seatunnel-e2e/seatunnel-flink-e2e/src/test/resources/assertion/fakesource_to_assert.conf:
##########
@@ -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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ field_name = "name,age"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/source-plugins/Fake
+}
+
+transform {
+ sql {
+ sql = "select name,age from fake"
+ }
+
+ # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+ # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}
+
+sink {
+ AssertSink {
+ rules =
+ [{
+ field_name = name
+ field_type = string
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN_LENGTH
+ rule_value = 3
+ },
+ {
+ rule_type = MAX_LENGTH
+ rule_value = 5
+ }
+ ]
+ },{
+ field_name = age
+ field_type = int
+ field_value = [
+ {
+ rule_type = NOT_NULL
+ },
+ {
+ rule_type = MIN
+ rule_value = 10
+ },
+ {
+ rule_type = MAX
+ rule_value = 20
+ }
+ ]
+ }
Review Comment:
Can you tell me where can i see the IT error?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ruanwenjun commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
ruanwenjun commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1164111389
@lhyundeadsoul Is there any count rule? e.g. I want to assert the batch row count is 4.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
Hisoka-X commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1171005418
> > > Both console and assert e2e IT failed. `org.apache.seatunnel.e2e.flink.fake.FakeSourceToConsoleIT`
> > > ```
> > > 22/06/30 17:18:28 ERROR FlinkContainer: Error: Could not find or load main class org.apache.seatunnel.core.flink.FlinkEnvParameterParser
> > >
> > >
> > > java.lang.AssertionError:
> > > Expected :0
> > > Actual :1
> > > <Click to see difference>
> > > ```
> >
> >
> > In your local?
> > Maybe it's in my local.
>
> I find the root cause : There is no `org.apache.seatunnel.core.flink.FlinkEnvParameterParser` in my target jar. and an error often occur while execute 'mvn install'
>
> > Failure to find org.apache.seatunnel:seatunnel:pom:${revision} in https://repo.maven.apache.org/maven2 was cached in the local repository, resolution will not be reattempted until the update interval of central has elapsed or updates are forced
>
> Have you ever had this?
Yes, also confusion meðŸ˜
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on a diff in pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on code in PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#discussion_r910936440
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
I have added AssertSink in whetherExecute
##########
seatunnel-connectors/seatunnel-connectors-flink/seatunnel-connector-flink-assert/src/main/java/org/apache/seatunnel/flink/assertion/sink/AssertSink.java:
##########
@@ -59,7 +61,7 @@ public void outputBatch(FlinkEnvironment env, DataSet<Row> inDataSet) {
throw new IllegalStateException("row :" + row + " fail rule: " + failRule);
});
return null;
- });
+ }).print();
Review Comment:
I have added AssertSink in whetherExecute to fix.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] lhyundeadsoul commented on pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
lhyundeadsoul commented on PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036#issuecomment-1160205272
PTAL @ruanwenjun Thx!
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] Hisoka-X merged pull request #2036: [Feature][Connector] add IT for Assert Sink in e2e module
Posted by GitBox <gi...@apache.org>.
Hisoka-X merged PR #2036:
URL: https://github.com/apache/incubator-seatunnel/pull/2036
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org