[jira] [Work logged] (BEAM-4681) Integrate support for timers using the portability APIs into Flink
[
https://issues.apache.org/jira/browse/BEAM-4681?focusedWorklogId=179609&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179609
]
ASF GitHub Bot logged work on BEAM-4681:
Author: ASF GitHub Bot
Created on: 29/Dec/18 04:04
Start Date: 29/Dec/18 04:04
Worklog Time Spent: 10m
Work Description: tweise commented on pull request #7359: [BEAM-4681]
Address synchronization issue for portable timers
URL: https://github.com/apache/beam/pull/7359#discussion_r29305
##
File path:
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/utils/NoopLock.java
##
@@ -0,0 +1,72 @@
+/*
+ * 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.beam.runners.flink.translation.utils;
+
+import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import javax.annotation.Nonnull;
+
+/**
+ * A lock which can always be acquired. It should not be used when a proper
lock is required, but it
+ * is useful as a performance optimization when locking is not necessary but
the code paths have to
+ * be shared between the locking and the non-locking variant.
+ *
+ * For example, in {@link
+ *
org.apache.beam.runners.flink.translation.wrappers.streaming.ExecutableStageDoFnOperator},
the
+ * locking on the state backend is only required when both timers and state
are used.
+ */
+public class NoopLock implements Lock, Serializable {
Review comment:
Since this may be useful outside of the Flink runner, maybe move it to a
common package like `org.apache.beam.sdk.util`?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179609)
Time Spent: 25h 50m (was: 25h 40m)
> Integrate support for timers using the portability APIs into Flink
> --
>
> Key: BEAM-4681
> URL: https://issues.apache.org/jira/browse/BEAM-4681
> Project: Beam
> Issue Type: Sub-task
> Components: runner-flink
>Reporter: Luke Cwik
>Assignee: Maximilian Michels
>Priority: Major
> Labels: portability, portability-flink
> Fix For: 2.9.0
>
> Time Spent: 25h 50m
> Remaining Estimate: 0h
>
> Consider using the code produced in BEAM-4658 to support timers.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179604&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179604
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 29/Dec/18 03:32
Start Date: 29/Dec/18 03:32
Worklog Time Spent: 10m
Work Description: tweise commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r27704
##
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java
##
@@ -626,13 +650,20 @@ public void testSourceWithNoReaderDoesNotShutdown()
throws Exception {
try {
thread.start();
-List>>
localReaders =
-sourceWrapper.getLocalReaders();
-while (localReaders != null && !localReaders.isEmpty()) {
+// Wait to see if the wrapper shuts down immediately in case it
doesn't have readers
Review comment:
Ideally we would check for the desired state instead of the assumption that
isRunning would flip within 200ms. But short of checking that `finalizeSource`
is in the run threads call stack I cannot see any way of doing that.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179604)
Time Spent: 2h (was: 1h 50m)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 2h
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179602&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179602
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 29/Dec/18 03:18
Start Date: 29/Dec/18 03:18
Worklog Time Spent: 10m
Work Description: tweise commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r26660
##
File path:
runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapper.java
##
@@ -464,16 +472,32 @@ private void
setNextWatermarkTimer(StreamingRuntimeContext runtime) {
/** Visible so that we can check this in tests. Must not be used for
anything else. */
@VisibleForTesting
- public List>
getLocalSplitSources() {
+ List>
getLocalSplitSources() {
return localSplitSources;
}
/** Visible so that we can check this in tests. Must not be used for
anything else. */
@VisibleForTesting
- public List> getLocalReaders() {
+ List> getLocalReaders() {
return localReaders;
}
+ /** Visible so that we can check this in tests. Must not be used for
anything else. */
+ @VisibleForTesting
+ boolean isRunning() {
+return isRunning;
+ }
+
+ /**
Review comment:
I would prefer we use `Whitebox::setInternalState` instead of adding an
extra method. Same could be done for `isRunning` above.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179602)
Time Spent: 1h 50m (was: 1h 40m)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 1h 50m
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-6284) [FLAKE][beam_PostCommit_Java_ValidatesRunner_Dataflow] TestRunner fails with result UNKNOWN on succeeded job and checks passed
[ https://issues.apache.org/jira/browse/BEAM-6284?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730505#comment-16730505 ] Andrew Pilloud commented on BEAM-6284: -- https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/1751/ > [FLAKE][beam_PostCommit_Java_ValidatesRunner_Dataflow] TestRunner fails with > result UNKNOWN on succeeded job and checks passed > -- > > Key: BEAM-6284 > URL: https://issues.apache.org/jira/browse/BEAM-6284 > Project: Beam > Issue Type: Bug > Components: test-failures, testing >Reporter: Mikhail Gryzykhin >Assignee: Jason Kuster >Priority: Major > Labels: currently-failing > > _Use this form to file an issue for test failure:_ > * > https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/testReport/junit/org.apache.beam.sdk.transforms/ViewTest/testWindowedSideInputFixedToGlobal/ > Initial investigation: > According to logs all test-relevant checks have passed and it seem to be > testing framework failure. > > _After you've filled out the above details, please [assign the issue to an > individual|https://beam.apache.org/contribute/postcommits-guides/index.html#find_specialist]. > Assignee should [treat test failures as > high-priority|https://beam.apache.org/contribute/postcommits-policies/#assigned-failing-test], > helping to fix the issue or find a more appropriate owner. See [Apache Beam > Post-Commit > Policies|https://beam.apache.org/contribute/postcommits-policies]._ -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-6310) org.apache.beam.sdk.io.gcp.bigquery.BigQueryNestedRecordsIT.testNestedRecords flaky
[
https://issues.apache.org/jira/browse/BEAM-6310?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730502#comment-16730502
]
Andrew Pilloud commented on BEAM-6310:
--
https://builds.apache.org/job/beam_PostCommit_Java/2226/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryNestedRecordsIT/testNestedRecords/
>
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryNestedRecordsIT.testNestedRecords
> flaky
>
>
> Key: BEAM-6310
> URL: https://issues.apache.org/jira/browse/BEAM-6310
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Reporter: Andrew Pilloud
>Assignee: Pablo Estrada
>Priority: Major
>
> https://builds.apache.org/job/beam_PostCommit_Java/2208/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryNestedRecordsIT/testNestedRecords/
> {code}
> com.google.api.client.googleapis.json.GoogleJsonResponseException: 404 Not
> Found
> {
> "code" : 404,
> "errors" : [ {
> "domain" : "global",
> "message" : "Not found: Dataset
> apache-beam-testing:_dataflow_temporary_dataset_78064",
> "reason" : "notFound"
> } ],
> "message" : "Not found: Dataset
> apache-beam-testing:_dataflow_temporary_dataset_78064"
> }
> at
> com.google.api.client.googleapis.json.GoogleJsonResponseException.from(GoogleJsonResponseException.java:150)
> at
> com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:113)
> at
> com.google.api.client.googleapis.services.json.AbstractGoogleJsonClientRequest.newExceptionOnError(AbstractGoogleJsonClientRequest.java:40)
> at
> com.google.api.client.googleapis.services.AbstractGoogleClientRequest$1.interceptResponse(AbstractGoogleClientRequest.java:401)
> at com.google.api.client.http.HttpRequest.execute(HttpRequest.java:1097)
> at
> com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:499)
> at
> com.google.api.client.googleapis.services.AbstractGoogleClientRequest.executeUnparsed(AbstractGoogleClientRequest.java:432)
> at
> com.google.api.client.googleapis.services.AbstractGoogleClientRequest.execute(AbstractGoogleClientRequest.java:549)
> at
> org.apache.beam.sdk.io.gcp.testing.BigqueryClient.queryUnflattened(BigqueryClient.java:302)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryNestedRecordsIT.runPipeline(BigQueryNestedRecordsIT.java:141)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryNestedRecordsIT.testNestedRecords(BigQueryNestedRecordsIT.java:74)
> 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.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.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:349)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:314)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:312)
> at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:292)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:396)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
> at
> org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
> at
> org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProc
[jira] [Commented] (BEAM-5415) [beam_PostCommit_Python_Verify] test is failing
[ https://issues.apache.org/jira/browse/BEAM-5415?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730493#comment-16730493 ] Valentyn Tymofieiev commented on BEAM-5415: --- This bug was originally opened a flake, but we could not correlate the flake with a failing job ID from the logs. Looks like now logs give more information, so let's focus this Jira on a last failure with similar symptom, pointed out by [~apilloud], that has clear indication of a failing job. Job id: 2018-12-28_08_18_05-13278266228563257972 Logs have several messages of the form: E Missing required coder_id on grpc_port for -11; using deprecated fallback. undefined I No unique name set for transform -11 Then the test times out. Assigning to [~robertwb] who added the "Missing required coder_id" error message and is investigating a similar issue. > [beam_PostCommit_Python_Verify] test is failing > --- > > Key: BEAM-5415 > URL: https://issues.apache.org/jira/browse/BEAM-5415 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Batkhuyag Batsaikhan >Assignee: Valentyn Tymofieiev >Priority: Major > > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Python_Verify/6003/] > Most likely culprit PR: https://github.com/apache/beam/pull/6403 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-4726) Reduce ParDo per element Invoke overhead
[
https://issues.apache.org/jira/browse/BEAM-4726?focusedWorklogId=179589&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179589
]
ASF GitHub Bot logged work on BEAM-4726:
Author: ASF GitHub Bot
Created on: 28/Dec/18 23:01
Start Date: 28/Dec/18 23:01
Worklog Time Spent: 10m
Work Description: lostluck commented on issue #7355: [BEAM-4726] Add
arity specialization for calling and returns.
URL: https://github.com/apache/beam/pull/7355#issuecomment-450440353
This broke post commits since I don't check if a value is nil before type
asserting.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179589)
Time Spent: 9h 10m (was: 9h)
> Reduce ParDo per element Invoke overhead
>
>
> Key: BEAM-4726
> URL: https://issues.apache.org/jira/browse/BEAM-4726
> Project: Beam
> Issue Type: Sub-task
> Components: sdk-go
>Reporter: Robert Burke
>Assignee: Robert Burke
>Priority: Major
> Time Spent: 9h 10m
> Remaining Estimate: 0h
>
> Each call to invoke allocates a new args []interface{}, but the common case
> is to run the same ProcessElement function over and again. It should be
> possible to have a container struct to retain the args slice, and avoid
> recomputing the indices for where to assign parameters before calling the
> ProcessElementFn.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5415) [beam_PostCommit_Python_Verify] test times out with "Missing required coder_id" messages in Dataflow job logs.
[ https://issues.apache.org/jira/browse/BEAM-5415?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730494#comment-16730494 ] Valentyn Tymofieiev commented on BEAM-5415: --- cc: [~altay] > [beam_PostCommit_Python_Verify] test times out with "Missing required > coder_id" messages in Dataflow job logs. > -- > > Key: BEAM-5415 > URL: https://issues.apache.org/jira/browse/BEAM-5415 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Batkhuyag Batsaikhan >Assignee: Robert Bradshaw >Priority: Major > > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Python_Verify/6003/] > Most likely culprit PR: https://github.com/apache/beam/pull/6403 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-5415) [beam_PostCommit_Python_Verify] test times out with "Missing required coder_id" messages in Dataflow job logs.
[ https://issues.apache.org/jira/browse/BEAM-5415?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Valentyn Tymofieiev updated BEAM-5415: -- Summary: [beam_PostCommit_Python_Verify] test times out with "Missing required coder_id" messages in Dataflow job logs. (was: [beam_PostCommit_Python_Verify] test is failing) > [beam_PostCommit_Python_Verify] test times out with "Missing required > coder_id" messages in Dataflow job logs. > -- > > Key: BEAM-5415 > URL: https://issues.apache.org/jira/browse/BEAM-5415 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Batkhuyag Batsaikhan >Assignee: Valentyn Tymofieiev >Priority: Major > > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Python_Verify/6003/] > Most likely culprit PR: https://github.com/apache/beam/pull/6403 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Assigned] (BEAM-5415) [beam_PostCommit_Python_Verify] test times out with "Missing required coder_id" messages in Dataflow job logs.
[ https://issues.apache.org/jira/browse/BEAM-5415?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Valentyn Tymofieiev reassigned BEAM-5415: - Assignee: Robert Bradshaw (was: Valentyn Tymofieiev) > [beam_PostCommit_Python_Verify] test times out with "Missing required > coder_id" messages in Dataflow job logs. > -- > > Key: BEAM-5415 > URL: https://issues.apache.org/jira/browse/BEAM-5415 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Batkhuyag Batsaikhan >Assignee: Robert Bradshaw >Priority: Major > > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Python_Verify/6003/] > Most likely culprit PR: https://github.com/apache/beam/pull/6403 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-6316) beam-sdks-python:portableWordCountBatch flaky: could not load docker image
[
https://issues.apache.org/jira/browse/BEAM-6316?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Maximilian Michels resolved BEAM-6316.
--
Resolution: Fixed
Fix Version/s: Not applicable
Thanks for reporting this.
> beam-sdks-python:portableWordCountBatch flaky: could not load docker image
> --
>
> Key: BEAM-6316
> URL: https://issues.apache.org/jira/browse/BEAM-6316
> Project: Beam
> Issue Type: Bug
> Components: runner-flink, test-failures
>Reporter: Andrew Pilloud
>Assignee: Maximilian Michels
>Priority: Major
> Fix For: Not applicable
>
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/105/
> {code}
> 10:08:30 INFO:root:Using latest locally built Python SDK docker image.
> 10:08:31 Unable to find image
> 'jenkins-docker-apache.bintray.io/beam/flink-job-server:latest' locally
> 10:08:31 INFO:root:Parsing unknown args: ['--environment_cache_millis=1']
> 10:08:31 docker: Error response from daemon: unknown: Repo 'apache' was not
> found.
> 10:08:31 See 'docker run --help'.
> 12:00:13 Build timed out (after 120 minutes). Marking the build as aborted.
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6316) beam-sdks-python:portableWordCountBatch flaky: could not load docker image
[
https://issues.apache.org/jira/browse/BEAM-6316?focusedWorklogId=179588&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179588
]
ASF GitHub Bot logged work on BEAM-6316:
Author: ASF GitHub Bot
Created on: 28/Dec/18 22:49
Start Date: 28/Dec/18 22:49
Worklog Time Spent: 10m
Work Description: mxm commented on pull request #7377: [BEAM-6316] Fix
container image name for PreCommit PortableWordCount
URL: https://github.com/apache/beam/pull/7377
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179588)
Time Spent: 0.5h (was: 20m)
> beam-sdks-python:portableWordCountBatch flaky: could not load docker image
> --
>
> Key: BEAM-6316
> URL: https://issues.apache.org/jira/browse/BEAM-6316
> Project: Beam
> Issue Type: Bug
> Components: runner-flink, test-failures
>Reporter: Andrew Pilloud
>Assignee: Maximilian Michels
>Priority: Major
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/105/
> {code}
> 10:08:30 INFO:root:Using latest locally built Python SDK docker image.
> 10:08:31 Unable to find image
> 'jenkins-docker-apache.bintray.io/beam/flink-job-server:latest' locally
> 10:08:31 INFO:root:Parsing unknown args: ['--environment_cache_millis=1']
> 10:08:31 docker: Error response from daemon: unknown: Repo 'apache' was not
> found.
> 10:08:31 See 'docker run --help'.
> 12:00:13 Build timed out (after 120 minutes). Marking the build as aborted.
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-6325) Error message is printed as bytes when failing to cross-compile
[
https://issues.apache.org/jira/browse/BEAM-6325?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Robert Burke resolved BEAM-6325.
Resolution: Fixed
Fix Version/s: Not applicable
Thanks for the notice. Casting to string added for the out.
> Error message is printed as bytes when failing to cross-compile
> ---
>
> Key: BEAM-6325
> URL: https://issues.apache.org/jira/browse/BEAM-6325
> Project: Beam
> Issue Type: Bug
> Components: sdk-go
>Reporter: Ivar Gaitan
>Assignee: Robert Burke
>Priority: Minor
> Fix For: Not applicable
>
> Time Spent: 40m
> Remaining Estimate: 0h
>
> In {{beam/runners/universal/runnerlib/compile.go}}:
> If cross-compile fails in {{BuildWorkerBinary()}} the error message is
> printed as bytes. This is because {{Cmd.CombinedOutput()}} returns a byte
> slice as its output, which is then formatted as it's default value.
> {{if out, err := cmd.CombinedOutput(); err != nil {}}
> {{ return fmt.Errorf("failed to cross-compile %v: %v\n%v", program, err,
> out)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[ https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179587&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179587 ] ASF GitHub Bot logged work on BEAM-5112: Author: ASF GitHub Bot Created on: 28/Dec/18 22:32 Start Date: 28/Dec/18 22:32 Worklog Time Spent: 10m Work Description: apilloud commented on pull request #6417: [BEAM-5112] Use Calcite codegen to implement BeamCalcRel URL: https://github.com/apache/beam/pull/6417 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179587) Time Spent: 5.5h (was: 5h 20m) > Investigate if Calcite can generate functions that we need > -- > > Key: BEAM-5112 > URL: https://issues.apache.org/jira/browse/BEAM-5112 > Project: Beam > Issue Type: Sub-task > Components: dsl-sql >Reporter: Rui Wang >Assignee: Andrew Pilloud >Priority: Major > Time Spent: 5.5h > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6325) Error message is printed as bytes when failing to cross-compile
[
https://issues.apache.org/jira/browse/BEAM-6325?focusedWorklogId=179583&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179583
]
ASF GitHub Bot logged work on BEAM-6325:
Author: ASF GitHub Bot
Created on: 28/Dec/18 22:17
Start Date: 28/Dec/18 22:17
Worklog Time Spent: 10m
Work Description: apilloud commented on issue #7375: [BEAM-6325] Cast
cross compile output from []byte to string for printing
URL: https://github.com/apache/beam/pull/7375#issuecomment-450436168
LGTM.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179583)
Time Spent: 0.5h (was: 20m)
> Error message is printed as bytes when failing to cross-compile
> ---
>
> Key: BEAM-6325
> URL: https://issues.apache.org/jira/browse/BEAM-6325
> Project: Beam
> Issue Type: Bug
> Components: sdk-go
>Reporter: Ivar Gaitan
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> In {{beam/runners/universal/runnerlib/compile.go}}:
> If cross-compile fails in {{BuildWorkerBinary()}} the error message is
> printed as bytes. This is because {{Cmd.CombinedOutput()}} returns a byte
> slice as its output, which is then formatted as it's default value.
> {{if out, err := cmd.CombinedOutput(); err != nil {}}
> {{ return fmt.Errorf("failed to cross-compile %v: %v\n%v", program, err,
> out)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-4725) Reduce framework overhead in Go SDK
[ https://issues.apache.org/jira/browse/BEAM-4725?focusedWorklogId=179586&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179586 ] ASF GitHub Bot logged work on BEAM-4725: Author: ASF GitHub Bot Created on: 28/Dec/18 22:26 Start Date: 28/Dec/18 22:26 Worklog Time Spent: 10m Work Description: lostluck commented on issue #7357: [BEAM-4725] Use unsafe to avoid small allocations to the heap. URL: https://github.com/apache/beam/pull/7357#issuecomment-450437077 You're not wrong. The problem is the indirection through the interfaces. *anything* could be in there and it won't be known necessarily at runtime. At least, it won't be known at package compile time, since it's a subsequent package that would use the function. The only way I can see around it, would be to do whole program optimization, and check the status of all parameters for all implementations of a given interface method. If nothing ever escapes, then the compiler can smooth things over. However, then it would still fail if one of the io.Read methods do cause their parameters to escape, and we're back where we started. As it stands, I don't think Go currently does a whole program optimization pass. Maybe someday though. You can learn more about some of the flaws in the escape analyzer (at least as of Go1.10) here: https://www.ardanlabs.com/blog/2018/01/escape-analysis-flaws.html This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179586) Time Spent: 50m (was: 40m) > Reduce framework overhead in Go SDK > --- > > Key: BEAM-4725 > URL: https://issues.apache.org/jira/browse/BEAM-4725 > Project: Beam > Issue Type: Improvement > Components: sdk-go >Reporter: Robert Burke >Assignee: Robert Burke >Priority: Major > Time Spent: 50m > Remaining Estimate: 0h > > The Go SDK produces correct results, but it also does redundant > recomputations, or allocations. This Jira is to track performance > improvements along any per-element path, where that overhead could be > significant. > Contributions should include at least one benchmark with before and after > values to demonstrate the value of the change. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-3612) Make it easy to generate type-specialized Go SDK reflectx.Funcs
[ https://issues.apache.org/jira/browse/BEAM-3612?focusedWorklogId=179585&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179585 ] ASF GitHub Bot logged work on BEAM-3612: Author: ASF GitHub Bot Created on: 28/Dec/18 22:18 Start Date: 28/Dec/18 22:18 Worklog Time Spent: 10m Work Description: apilloud commented on issue #7361: [BEAM-3612] Generate type assertion shims for beam. URL: https://github.com/apache/beam/pull/7361#issuecomment-450436247 Run Go PreCommit This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179585) Time Spent: 8h 10m (was: 8h) > Make it easy to generate type-specialized Go SDK reflectx.Funcs > --- > > Key: BEAM-3612 > URL: https://issues.apache.org/jira/browse/BEAM-3612 > Project: Beam > Issue Type: Improvement > Components: sdk-go >Reporter: Henning Rohde >Assignee: Robert Burke >Priority: Major > Time Spent: 8h 10m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6325) Error message is printed as bytes when failing to cross-compile
[
https://issues.apache.org/jira/browse/BEAM-6325?focusedWorklogId=179584&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179584
]
ASF GitHub Bot logged work on BEAM-6325:
Author: ASF GitHub Bot
Created on: 28/Dec/18 22:17
Start Date: 28/Dec/18 22:17
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #7375: [BEAM-6325]
Cast cross compile output from []byte to string for printing
URL: https://github.com/apache/beam/pull/7375
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179584)
Time Spent: 40m (was: 0.5h)
> Error message is printed as bytes when failing to cross-compile
> ---
>
> Key: BEAM-6325
> URL: https://issues.apache.org/jira/browse/BEAM-6325
> Project: Beam
> Issue Type: Bug
> Components: sdk-go
>Reporter: Ivar Gaitan
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 40m
> Remaining Estimate: 0h
>
> In {{beam/runners/universal/runnerlib/compile.go}}:
> If cross-compile fails in {{BuildWorkerBinary()}} the error message is
> printed as bytes. This is because {{Cmd.CombinedOutput()}} returns a byte
> slice as its output, which is then formatted as it's default value.
> {{if out, err := cmd.CombinedOutput(); err != nil {}}
> {{ return fmt.Errorf("failed to cross-compile %v: %v\n%v", program, err,
> out)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-4725) Reduce framework overhead in Go SDK
[ https://issues.apache.org/jira/browse/BEAM-4725?focusedWorklogId=179582&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179582 ] ASF GitHub Bot logged work on BEAM-4725: Author: ASF GitHub Bot Created on: 28/Dec/18 22:17 Start Date: 28/Dec/18 22:17 Worklog Time Spent: 10m Work Description: apilloud commented on pull request #7357: [BEAM-4725] Use unsafe to avoid small allocations to the heap. URL: https://github.com/apache/beam/pull/7357 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179582) Time Spent: 40m (was: 0.5h) > Reduce framework overhead in Go SDK > --- > > Key: BEAM-4725 > URL: https://issues.apache.org/jira/browse/BEAM-4725 > Project: Beam > Issue Type: Improvement > Components: sdk-go >Reporter: Robert Burke >Assignee: Robert Burke >Priority: Major > Time Spent: 40m > Remaining Estimate: 0h > > The Go SDK produces correct results, but it also does redundant > recomputations, or allocations. This Jira is to track performance > improvements along any per-element path, where that overhead could be > significant. > Contributions should include at least one benchmark with before and after > values to demonstrate the value of the change. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-4725) Reduce framework overhead in Go SDK
[ https://issues.apache.org/jira/browse/BEAM-4725?focusedWorklogId=179581&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179581 ] ASF GitHub Bot logged work on BEAM-4725: Author: ASF GitHub Bot Created on: 28/Dec/18 22:16 Start Date: 28/Dec/18 22:16 Worklog Time Spent: 10m Work Description: apilloud commented on issue #7357: [BEAM-4725] Use unsafe to avoid small allocations to the heap. URL: https://github.com/apache/beam/pull/7357#issuecomment-450436082 The assertion here is that Read/Write won't retain a reference to the data after they return. Sounds reasonable, surprised the go compiler doesn't make that assumption too. LGTM This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179581) Time Spent: 0.5h (was: 20m) > Reduce framework overhead in Go SDK > --- > > Key: BEAM-4725 > URL: https://issues.apache.org/jira/browse/BEAM-4725 > Project: Beam > Issue Type: Improvement > Components: sdk-go >Reporter: Robert Burke >Assignee: Robert Burke >Priority: Major > Time Spent: 0.5h > Remaining Estimate: 0h > > The Go SDK produces correct results, but it also does redundant > recomputations, or allocations. This Jira is to track performance > improvements along any per-element path, where that overhead could be > significant. > Contributions should include at least one benchmark with before and after > values to demonstrate the value of the change. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6319) BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
[
https://issues.apache.org/jira/browse/BEAM-6319?focusedWorklogId=179577&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179577
]
ASF GitHub Bot logged work on BEAM-6319:
Author: ASF GitHub Bot
Created on: 28/Dec/18 22:04
Start Date: 28/Dec/18 22:04
Worklog Time Spent: 10m
Work Description: apilloud commented on issue #7368:
[BEAM-6319][BEAM-6312][BEAM-6311] Disable BigQueryIo validation in
BigQueryToTableIT
URL: https://github.com/apache/beam/pull/7368#issuecomment-450434570
LGTM
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179577)
Time Spent: 0.5h (was: 20m)
> BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
> ---
>
> Key: BEAM-6319
> URL: https://issues.apache.org/jira/browse/BEAM-6319
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Reporter: Andrew Pilloud
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PostCommit_Java/2213/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryToTableIT/testNewTypesQueryWithoutReshuffleWithCustom/
> {code}
> java.lang.IllegalArgumentException: Validation of query "SELECT bytes, date,
> time FROM [apache-beam-testing:bq_query_to_table_1545892541359_421.types]"
> failed. If the query depends on an earlier stage of the pipeline, This
> validation can be disabled using #withoutValidation.
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$TypedRead.validate(BigQueryIO.java:705)
> at
> org.apache.beam.sdk.Pipeline$ValidateVisitor.enterCompositeTransform(Pipeline.java:641)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:645)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:311)
> at
> org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:245)
> at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
> at org.apache.beam.sdk.Pipeline.validate(Pipeline.java:577)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:312)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.runBigQueryToTablePipeline(BigQueryToTableIT.java:111)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom(BigQueryToTableIT.java:307)
> 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.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.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:349)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:314)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:312)
> at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:292)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:396)
> at
> org.gradle.api
[jira] [Work logged] (BEAM-6319) BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
[
https://issues.apache.org/jira/browse/BEAM-6319?focusedWorklogId=179578&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179578
]
ASF GitHub Bot logged work on BEAM-6319:
Author: ASF GitHub Bot
Created on: 28/Dec/18 22:04
Start Date: 28/Dec/18 22:04
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #7368:
[BEAM-6319][BEAM-6312][BEAM-6311] Disable BigQueryIo validation in
BigQueryToTableIT
URL: https://github.com/apache/beam/pull/7368
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179578)
Time Spent: 40m (was: 0.5h)
> BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
> ---
>
> Key: BEAM-6319
> URL: https://issues.apache.org/jira/browse/BEAM-6319
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Reporter: Andrew Pilloud
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 40m
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PostCommit_Java/2213/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryToTableIT/testNewTypesQueryWithoutReshuffleWithCustom/
> {code}
> java.lang.IllegalArgumentException: Validation of query "SELECT bytes, date,
> time FROM [apache-beam-testing:bq_query_to_table_1545892541359_421.types]"
> failed. If the query depends on an earlier stage of the pipeline, This
> validation can be disabled using #withoutValidation.
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$TypedRead.validate(BigQueryIO.java:705)
> at
> org.apache.beam.sdk.Pipeline$ValidateVisitor.enterCompositeTransform(Pipeline.java:641)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:645)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:311)
> at
> org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:245)
> at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
> at org.apache.beam.sdk.Pipeline.validate(Pipeline.java:577)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:312)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.runBigQueryToTablePipeline(BigQueryToTableIT.java:111)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom(BigQueryToTableIT.java:307)
> 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.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.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:349)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:314)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:312)
> at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:292)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:396)
> at
> org.gradle.api.internal.tasks.testi
[jira] [Work logged] (BEAM-6316) beam-sdks-python:portableWordCountBatch flaky: could not load docker image
[
https://issues.apache.org/jira/browse/BEAM-6316?focusedWorklogId=179576&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179576
]
ASF GitHub Bot logged work on BEAM-6316:
Author: ASF GitHub Bot
Created on: 28/Dec/18 22:02
Start Date: 28/Dec/18 22:02
Worklog Time Spent: 10m
Work Description: apilloud commented on issue #7377: [BEAM-6316] Fix
container image name for PreCommit PortableWordCount
URL: https://github.com/apache/beam/pull/7377#issuecomment-450434384
LGTM
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179576)
Time Spent: 20m (was: 10m)
> beam-sdks-python:portableWordCountBatch flaky: could not load docker image
> --
>
> Key: BEAM-6316
> URL: https://issues.apache.org/jira/browse/BEAM-6316
> Project: Beam
> Issue Type: Bug
> Components: runner-flink, test-failures
>Reporter: Andrew Pilloud
>Assignee: Maximilian Michels
>Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/105/
> {code}
> 10:08:30 INFO:root:Using latest locally built Python SDK docker image.
> 10:08:31 Unable to find image
> 'jenkins-docker-apache.bintray.io/beam/flink-job-server:latest' locally
> 10:08:31 INFO:root:Parsing unknown args: ['--environment_cache_millis=1']
> 10:08:31 docker: Error response from daemon: unknown: Repo 'apache' was not
> found.
> 10:08:31 See 'docker run --help'.
> 12:00:13 Build timed out (after 120 minutes). Marking the build as aborted.
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[ https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179575&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179575 ] ASF GitHub Bot logged work on BEAM-5112: Author: ASF GitHub Bot Created on: 28/Dec/18 22:01 Start Date: 28/Dec/18 22:01 Worklog Time Spent: 10m Work Description: apilloud commented on issue #6417: [BEAM-5112] Use Calcite codegen to implement BeamCalcRel URL: https://github.com/apache/beam/pull/6417#issuecomment-450434159 This is a 12% speedup in a 20 second run and 22% in 60 second run of Nexmark Query 1. It was a significant speedup on a ~30 minute run, but it doesn't look like we have one of those in our postcommit anymore. Jobs that are very short or lite on Calc logic might see a slowdown. There is probably some opportunity for optimization still. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179575) Time Spent: 5h 20m (was: 5h 10m) > Investigate if Calcite can generate functions that we need > -- > > Key: BEAM-5112 > URL: https://issues.apache.org/jira/browse/BEAM-5112 > Project: Beam > Issue Type: Sub-task > Components: dsl-sql >Reporter: Rui Wang >Assignee: Andrew Pilloud >Priority: Major > Time Spent: 5h 20m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-6319) BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
[
https://issues.apache.org/jira/browse/BEAM-6319?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730472#comment-16730472
]
Boyuan Zhang commented on BEAM-6319:
Took another look at this error. It's true that BigQueryIO.wirte() should run
without validation since output table is written by the pipeline. But another
part of this error is, the dataset and table is not created properly before
pipeline starting.
> BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
> ---
>
> Key: BEAM-6319
> URL: https://issues.apache.org/jira/browse/BEAM-6319
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Reporter: Andrew Pilloud
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PostCommit_Java/2213/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryToTableIT/testNewTypesQueryWithoutReshuffleWithCustom/
> {code}
> java.lang.IllegalArgumentException: Validation of query "SELECT bytes, date,
> time FROM [apache-beam-testing:bq_query_to_table_1545892541359_421.types]"
> failed. If the query depends on an earlier stage of the pipeline, This
> validation can be disabled using #withoutValidation.
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$TypedRead.validate(BigQueryIO.java:705)
> at
> org.apache.beam.sdk.Pipeline$ValidateVisitor.enterCompositeTransform(Pipeline.java:641)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:645)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:311)
> at
> org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:245)
> at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
> at org.apache.beam.sdk.Pipeline.validate(Pipeline.java:577)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:312)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.runBigQueryToTablePipeline(BigQueryToTableIT.java:111)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom(BigQueryToTableIT.java:307)
> 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.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.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:349)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:314)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:312)
> at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:292)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:396)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
> at
> org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
> at
> org.gradle.api.internal.tasks.testing.SuiteTestCla
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179574&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179574
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 21:41
Start Date: 28/Dec/18 21:41
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244413151
##
File path:
sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/integrationtest/BeamSqlDateFunctionsIntegrationTest.java
##
@@ -55,11 +56,13 @@ public Void apply(Iterable input) {
Row row = iter.next();
// LOCALTIME
DateTime date = DateTime.now();
- assertTrue(date.getMillis() - row.getDateTime(0).getMillis() < 1000);
- assertTrue(date.getMillis() - row.getDateTime(1).getMillis() < 1000);
- assertTrue(date.getMillis() - row.getDateTime(2).getMillis() < 1000);
- assertTrue(date.getMillis() - row.getDateTime(3).getMillis() < 1000);
- assertTrue(date.getMillis() - row.getDateTime(4).getMillis() < 1000);
+ long millis = date.getMillis();
+ int timeMillis = (int) (date.getMillis() % MILLIS_PER_DAY);
+ assertTrue(timeMillis - row.getDateTime(0).getMillis() < 1000);
+ assertTrue(millis - row.getDateTime(1).getMillis() < 1000);
+ assertTrue(millis - row.getDateTime(2).getMillis() < (MILLIS_PER_DAY +
1000));
Review comment:
This is a bad test. I've cleaned it up significantly.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179574)
Time Spent: 5h 10m (was: 5h)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 5h 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179573&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179573
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 21:39
Start Date: 28/Dec/18 21:39
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244412810
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -96,43 +224,243 @@ public boolean isInputSortRelAndLimitOnly() {
/** {@code CalcFn} is the executor for a {@link BeamCalcRel} step. */
private static class CalcFn extends DoFn {
-private BeamSqlExpressionExecutor executor;
-private Schema outputSchema;
+private final String processElementBlock;
+private final Schema outputSchema;
+private transient @Nullable ScriptEvaluator se = null;
-public CalcFn(BeamSqlExpressionExecutor executor, Schema outputSchema) {
- super();
- this.executor = executor;
+public CalcFn(String processElementBlock, Schema outputSchema) {
+ this.processElementBlock = processElementBlock;
this.outputSchema = outputSchema;
}
+ScriptEvaluator compile() {
+ ScriptEvaluator se = new ScriptEvaluator();
+ se.setParameters(
+ new String[] {outputSchemaParam.name, processContextParam.name,
DataContext.ROOT.name},
+ new Class[] {
+(Class) outputSchemaParam.getType(),
+(Class) processContextParam.getType(),
+(Class) DataContext.ROOT.getType()
+ });
+ try {
+se.cook(processElementBlock);
+ } catch (CompileException e) {
+throw new RuntimeException("Could not compile CalcFn: " +
processElementBlock, e);
+ }
+ return se;
+}
+
@Setup
public void setup() {
- executor.prepare();
+ this.se = compile();
}
@ProcessElement
public void processElement(ProcessContext c) {
- Row inputRow = c.element();
- @Nullable
- List rawResultValues =
- executor.execute(inputRow, null,
BeamSqlExpressionEnvironments.forRow(inputRow, null));
-
- if (rawResultValues != null) {
-List castResultValues =
-IntStream.range(0, outputSchema.getFieldCount())
-.mapToObj(i -> castField(rawResultValues, i))
-.collect(toList());
-
c.output(Row.withSchema(outputSchema).addValues(castResultValues).build());
+ assert se != null;
+ try {
+se.evaluate(new Object[] {outputSchema, c, CONTEXT_INSTANCE});
+ } catch (InvocationTargetException e) {
+throw new RuntimeException(
+"CalcFn failed to evaluate: " + processElementBlock, e.getCause());
+ }
+}
+ }
+
+ private static final Map rawTypeMap =
+ ImmutableMap.builder()
+ .put(TypeName.BYTE, Byte.class)
+ .put(TypeName.INT16, Short.class)
+ .put(TypeName.INT32, Integer.class)
+ .put(TypeName.INT64, Long.class)
+ .put(TypeName.FLOAT, Float.class)
+ .put(TypeName.DOUBLE, Double.class)
+ .build();
+
+ private Expression castOutput(Expression value, FieldType toType) {
+if (value.getType() == Object.class) {
+ // fast copy path, just pass object through
+ return value;
+} else if (toType.getTypeName().isDateType()
Review comment:
Both done.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179573)
Time Spent: 5h (was: 4h 50m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179572&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179572
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 21:33
Start Date: 28/Dec/18 21:33
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244411820
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -96,43 +224,243 @@ public boolean isInputSortRelAndLimitOnly() {
/** {@code CalcFn} is the executor for a {@link BeamCalcRel} step. */
private static class CalcFn extends DoFn {
-private BeamSqlExpressionExecutor executor;
-private Schema outputSchema;
+private final String processElementBlock;
+private final Schema outputSchema;
+private transient @Nullable ScriptEvaluator se = null;
-public CalcFn(BeamSqlExpressionExecutor executor, Schema outputSchema) {
- super();
- this.executor = executor;
+public CalcFn(String processElementBlock, Schema outputSchema) {
+ this.processElementBlock = processElementBlock;
this.outputSchema = outputSchema;
}
+ScriptEvaluator compile() {
+ ScriptEvaluator se = new ScriptEvaluator();
+ se.setParameters(
+ new String[] {outputSchemaParam.name, processContextParam.name,
DataContext.ROOT.name},
+ new Class[] {
+(Class) outputSchemaParam.getType(),
+(Class) processContextParam.getType(),
+(Class) DataContext.ROOT.getType()
+ });
+ try {
+se.cook(processElementBlock);
+ } catch (CompileException e) {
+throw new RuntimeException("Could not compile CalcFn: " +
processElementBlock, e);
+ }
+ return se;
+}
+
@Setup
public void setup() {
- executor.prepare();
+ this.se = compile();
}
@ProcessElement
public void processElement(ProcessContext c) {
- Row inputRow = c.element();
- @Nullable
- List rawResultValues =
- executor.execute(inputRow, null,
BeamSqlExpressionEnvironments.forRow(inputRow, null));
-
- if (rawResultValues != null) {
-List castResultValues =
-IntStream.range(0, outputSchema.getFieldCount())
-.mapToObj(i -> castField(rawResultValues, i))
-.collect(toList());
-
c.output(Row.withSchema(outputSchema).addValues(castResultValues).build());
+ assert se != null;
+ try {
+se.evaluate(new Object[] {outputSchema, c, CONTEXT_INSTANCE});
+ } catch (InvocationTargetException e) {
+throw new RuntimeException(
+"CalcFn failed to evaluate: " + processElementBlock, e.getCause());
+ }
+}
+ }
+
+ private static final Map rawTypeMap =
+ ImmutableMap.builder()
+ .put(TypeName.BYTE, Byte.class)
+ .put(TypeName.INT16, Short.class)
+ .put(TypeName.INT32, Integer.class)
+ .put(TypeName.INT64, Long.class)
+ .put(TypeName.FLOAT, Float.class)
+ .put(TypeName.DOUBLE, Double.class)
+ .build();
+
+ private Expression castOutput(Expression value, FieldType toType) {
+if (value.getType() == Object.class) {
+ // fast copy path, just pass object through
+ return value;
+} else if (toType.getTypeName().isDateType()
+&& value.getType() instanceof Class
+&& !Types.isAssignableFrom(ReadableInstant.class, (Class)
value.getType())) {
+ Expression valueDateTime = value;
+ if (toType.getMetadata() == null) {
+if (value.getType() == java.sql.Timestamp.class) {
Review comment:
The code generator only converts to the `java.sql` types if there are UDFs,
otherwise they will remain primitive types. So we don't always need to convert
back.
I am comparing metadata because `equals` compares things I don't care about
here, like nullability. Looks like there is a new `typesEqual` helper I can use
here.
I don't think this logic is compatible with a `switch/case` format.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179572)
Time Spent: 4h 50m (was: 4h 40m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/ji
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[ https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179571&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179571 ] ASF GitHub Bot logged work on BEAM-5112: Author: ASF GitHub Bot Created on: 28/Dec/18 21:14 Start Date: 28/Dec/18 21:14 Worklog Time Spent: 10m Work Description: akedin commented on issue #6417: [BEAM-5112] Use Calcite codegen to implement BeamCalcRel URL: https://github.com/apache/beam/pull/6417#issuecomment-450428567 And what's the perf change after this? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179571) Time Spent: 4h 40m (was: 4.5h) > Investigate if Calcite can generate functions that we need > -- > > Key: BEAM-5112 > URL: https://issues.apache.org/jira/browse/BEAM-5112 > Project: Beam > Issue Type: Sub-task > Components: dsl-sql >Reporter: Rui Wang >Assignee: Andrew Pilloud >Priority: Major > Time Spent: 4h 40m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179570&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179570
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 21:13
Start Date: 28/Dec/18 21:13
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244408908
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -96,43 +224,243 @@ public boolean isInputSortRelAndLimitOnly() {
/** {@code CalcFn} is the executor for a {@link BeamCalcRel} step. */
private static class CalcFn extends DoFn {
-private BeamSqlExpressionExecutor executor;
-private Schema outputSchema;
+private final String processElementBlock;
+private final Schema outputSchema;
+private transient @Nullable ScriptEvaluator se = null;
-public CalcFn(BeamSqlExpressionExecutor executor, Schema outputSchema) {
- super();
- this.executor = executor;
+public CalcFn(String processElementBlock, Schema outputSchema) {
+ this.processElementBlock = processElementBlock;
this.outputSchema = outputSchema;
}
+ScriptEvaluator compile() {
+ ScriptEvaluator se = new ScriptEvaluator();
+ se.setParameters(
+ new String[] {outputSchemaParam.name, processContextParam.name,
DataContext.ROOT.name},
+ new Class[] {
+(Class) outputSchemaParam.getType(),
+(Class) processContextParam.getType(),
+(Class) DataContext.ROOT.getType()
+ });
+ try {
+se.cook(processElementBlock);
+ } catch (CompileException e) {
+throw new RuntimeException("Could not compile CalcFn: " +
processElementBlock, e);
+ }
+ return se;
+}
+
@Setup
public void setup() {
- executor.prepare();
+ this.se = compile();
}
@ProcessElement
public void processElement(ProcessContext c) {
- Row inputRow = c.element();
- @Nullable
- List rawResultValues =
- executor.execute(inputRow, null,
BeamSqlExpressionEnvironments.forRow(inputRow, null));
-
- if (rawResultValues != null) {
-List castResultValues =
-IntStream.range(0, outputSchema.getFieldCount())
-.mapToObj(i -> castField(rawResultValues, i))
-.collect(toList());
-
c.output(Row.withSchema(outputSchema).addValues(castResultValues).build());
+ assert se != null;
+ try {
+se.evaluate(new Object[] {outputSchema, c, CONTEXT_INSTANCE});
+ } catch (InvocationTargetException e) {
+throw new RuntimeException(
+"CalcFn failed to evaluate: " + processElementBlock, e.getCause());
+ }
+}
+ }
+
+ private static final Map rawTypeMap =
+ ImmutableMap.builder()
+ .put(TypeName.BYTE, Byte.class)
+ .put(TypeName.INT16, Short.class)
+ .put(TypeName.INT32, Integer.class)
+ .put(TypeName.INT64, Long.class)
+ .put(TypeName.FLOAT, Float.class)
+ .put(TypeName.DOUBLE, Double.class)
+ .build();
+
+ private Expression castOutput(Expression value, FieldType toType) {
+if (value.getType() == Object.class) {
+ // fast copy path, just pass object through
+ return value;
+} else if (toType.getTypeName().isDateType()
+&& value.getType() instanceof Class
+&& !Types.isAssignableFrom(ReadableInstant.class, (Class)
value.getType())) {
Review comment:
No, it isn't. There are things that don't resolve to a type that come
through here. I believe constants are one example. I will change this so they
exit in the fast path too.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179570)
Time Spent: 4.5h (was: 4h 20m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 4.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-6168) Allow modification of JSON value before writing to ElasticSearch
[
https://issues.apache.org/jira/browse/BEAM-6168?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730468#comment-16730468
]
Mark Norkin commented on BEAM-6168:
---
[~echauchot] thank you for the reply.
The workaround you propose would probably work for me, it's just the downside
that this is some other job which should be maintained separately.
The point you made about behavior of other connectors, just to be clear, is it
the issue of breaking underlying connectors ? If so, I'm not sure how providing
mutation on beam level IO would break that as I'm assuming it can be performed
before actually going to underlying connector. Can you elaborate on that a bit ?
Thank you.
> Allow modification of JSON value before writing to ElasticSearch
>
>
> Key: BEAM-6168
> URL: https://issues.apache.org/jira/browse/BEAM-6168
> Project: Beam
> Issue Type: Improvement
> Components: io-java-elasticsearch
>Reporter: Mark Norkin
>Assignee: Etienne Chauchot
>Priority: Major
>
> I have an Apache Beam streaming job which reads data from Kafka and writes to
> ElasticSearch using ElasticSearchIO.
> The issue I'm having is that messages in Kafka already have _{{key}}_ field,
> and using {{ElasticSearchIO.Write.withIdFn()}} I'm mapping this field to
> document _{{_id}}_ field in ElasticSearch.
> Having a big volume of data I don't want the _{{key}}_ field to be also
> written to ElasticSearch as part of _{{_source}}_.
> Is there an option/workaround that would allow doing that?
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179569&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179569
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 21:08
Start Date: 28/Dec/18 21:08
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244408100
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -96,43 +224,243 @@ public boolean isInputSortRelAndLimitOnly() {
/** {@code CalcFn} is the executor for a {@link BeamCalcRel} step. */
private static class CalcFn extends DoFn {
-private BeamSqlExpressionExecutor executor;
-private Schema outputSchema;
+private final String processElementBlock;
+private final Schema outputSchema;
+private transient @Nullable ScriptEvaluator se = null;
-public CalcFn(BeamSqlExpressionExecutor executor, Schema outputSchema) {
- super();
- this.executor = executor;
+public CalcFn(String processElementBlock, Schema outputSchema) {
+ this.processElementBlock = processElementBlock;
this.outputSchema = outputSchema;
}
+ScriptEvaluator compile() {
+ ScriptEvaluator se = new ScriptEvaluator();
+ se.setParameters(
+ new String[] {outputSchemaParam.name, processContextParam.name,
DataContext.ROOT.name},
+ new Class[] {
+(Class) outputSchemaParam.getType(),
+(Class) processContextParam.getType(),
+(Class) DataContext.ROOT.getType()
+ });
+ try {
+se.cook(processElementBlock);
+ } catch (CompileException e) {
+throw new RuntimeException("Could not compile CalcFn: " +
processElementBlock, e);
+ }
+ return se;
+}
+
@Setup
public void setup() {
- executor.prepare();
+ this.se = compile();
}
@ProcessElement
public void processElement(ProcessContext c) {
- Row inputRow = c.element();
- @Nullable
- List rawResultValues =
- executor.execute(inputRow, null,
BeamSqlExpressionEnvironments.forRow(inputRow, null));
-
- if (rawResultValues != null) {
-List castResultValues =
-IntStream.range(0, outputSchema.getFieldCount())
-.mapToObj(i -> castField(rawResultValues, i))
-.collect(toList());
-
c.output(Row.withSchema(outputSchema).addValues(castResultValues).build());
+ assert se != null;
+ try {
+se.evaluate(new Object[] {outputSchema, c, CONTEXT_INSTANCE});
+ } catch (InvocationTargetException e) {
+throw new RuntimeException(
+"CalcFn failed to evaluate: " + processElementBlock, e.getCause());
+ }
+}
+ }
+
+ private static final Map rawTypeMap =
+ ImmutableMap.builder()
+ .put(TypeName.BYTE, Byte.class)
+ .put(TypeName.INT16, Short.class)
+ .put(TypeName.INT32, Integer.class)
+ .put(TypeName.INT64, Long.class)
+ .put(TypeName.FLOAT, Float.class)
+ .put(TypeName.DOUBLE, Double.class)
+ .build();
+
+ private Expression castOutput(Expression value, FieldType toType) {
+if (value.getType() == Object.class) {
+ // fast copy path, just pass object through
+ return value;
+} else if (toType.getTypeName().isDateType()
+&& value.getType() instanceof Class
+&& !Types.isAssignableFrom(ReadableInstant.class, (Class)
value.getType())) {
+ Expression valueDateTime = value;
+ if (toType.getMetadata() == null) {
+if (value.getType() == java.sql.Timestamp.class) {
+ valueDateTime =
Expressions.call(BuiltInMethod.TIMESTAMP_TO_LONG.method, valueDateTime);
+}
+ } else if (Arrays.equals(toType.getMetadata(),
CalciteUtils.TIME.getMetadata())) {
+if (value.getType() == java.sql.Time.class) {
+ valueDateTime = Expressions.call(BuiltInMethod.TIME_TO_INT.method,
valueDateTime);
+}
+ } else if (Arrays.equals(toType.getMetadata(),
CalciteUtils.DATE.getMetadata())) {
+if (value.getType() == java.sql.Date.class) {
+ valueDateTime = Expressions.call(BuiltInMethod.DATE_TO_INT.method,
valueDateTime);
+}
+valueDateTime = Expressions.multiply(valueDateTime,
Expressions.constant(MILLIS_PER_DAY));
+ } else {
+throw new IllegalArgumentException(
+"Unknown DateTime type " + new String(toType.getMetadata(),
UTF_8));
+ }
+ valueDateTime =
+ Expressions.new_(
+ DateTime.class,
+ valueDateTime,
+ Expressions.parameter(DateTimeZone.class,
"org.joda.time.DateTime
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179568&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179568
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 21:01
Start Date: 28/Dec/18 21:01
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244407083
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -96,43 +224,243 @@ public boolean isInputSortRelAndLimitOnly() {
/** {@code CalcFn} is the executor for a {@link BeamCalcRel} step. */
private static class CalcFn extends DoFn {
-private BeamSqlExpressionExecutor executor;
-private Schema outputSchema;
+private final String processElementBlock;
+private final Schema outputSchema;
+private transient @Nullable ScriptEvaluator se = null;
-public CalcFn(BeamSqlExpressionExecutor executor, Schema outputSchema) {
- super();
- this.executor = executor;
+public CalcFn(String processElementBlock, Schema outputSchema) {
+ this.processElementBlock = processElementBlock;
this.outputSchema = outputSchema;
}
+ScriptEvaluator compile() {
+ ScriptEvaluator se = new ScriptEvaluator();
+ se.setParameters(
+ new String[] {outputSchemaParam.name, processContextParam.name,
DataContext.ROOT.name},
+ new Class[] {
+(Class) outputSchemaParam.getType(),
+(Class) processContextParam.getType(),
+(Class) DataContext.ROOT.getType()
+ });
+ try {
+se.cook(processElementBlock);
+ } catch (CompileException e) {
+throw new RuntimeException("Could not compile CalcFn: " +
processElementBlock, e);
+ }
+ return se;
+}
+
@Setup
public void setup() {
- executor.prepare();
+ this.se = compile();
}
@ProcessElement
public void processElement(ProcessContext c) {
- Row inputRow = c.element();
- @Nullable
- List rawResultValues =
- executor.execute(inputRow, null,
BeamSqlExpressionEnvironments.forRow(inputRow, null));
-
- if (rawResultValues != null) {
-List castResultValues =
-IntStream.range(0, outputSchema.getFieldCount())
-.mapToObj(i -> castField(rawResultValues, i))
-.collect(toList());
-
c.output(Row.withSchema(outputSchema).addValues(castResultValues).build());
+ assert se != null;
+ try {
+se.evaluate(new Object[] {outputSchema, c, CONTEXT_INSTANCE});
+ } catch (InvocationTargetException e) {
+throw new RuntimeException(
+"CalcFn failed to evaluate: " + processElementBlock, e.getCause());
+ }
+}
+ }
+
+ private static final Map rawTypeMap =
+ ImmutableMap.builder()
+ .put(TypeName.BYTE, Byte.class)
+ .put(TypeName.INT16, Short.class)
+ .put(TypeName.INT32, Integer.class)
+ .put(TypeName.INT64, Long.class)
+ .put(TypeName.FLOAT, Float.class)
+ .put(TypeName.DOUBLE, Double.class)
+ .build();
+
+ private Expression castOutput(Expression value, FieldType toType) {
+if (value.getType() == Object.class) {
Review comment:
See the comment below. This is the fast copy path, which is actually used
quite a bit. When a field is copied from input to output without modification
code is generated to fetch it with `getValue` which returns an object. That
object can be passed through to the new row without any conversion.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179568)
Time Spent: 4h 10m (was: 4h)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 4h 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179567&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179567
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:54
Start Date: 28/Dec/18 20:54
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244406033
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -96,43 +224,243 @@ public boolean isInputSortRelAndLimitOnly() {
/** {@code CalcFn} is the executor for a {@link BeamCalcRel} step. */
private static class CalcFn extends DoFn {
-private BeamSqlExpressionExecutor executor;
-private Schema outputSchema;
+private final String processElementBlock;
+private final Schema outputSchema;
+private transient @Nullable ScriptEvaluator se = null;
-public CalcFn(BeamSqlExpressionExecutor executor, Schema outputSchema) {
- super();
- this.executor = executor;
+public CalcFn(String processElementBlock, Schema outputSchema) {
+ this.processElementBlock = processElementBlock;
this.outputSchema = outputSchema;
}
+ScriptEvaluator compile() {
+ ScriptEvaluator se = new ScriptEvaluator();
+ se.setParameters(
+ new String[] {outputSchemaParam.name, processContextParam.name,
DataContext.ROOT.name},
+ new Class[] {
+(Class) outputSchemaParam.getType(),
+(Class) processContextParam.getType(),
+(Class) DataContext.ROOT.getType()
+ });
+ try {
+se.cook(processElementBlock);
+ } catch (CompileException e) {
+throw new RuntimeException("Could not compile CalcFn: " +
processElementBlock, e);
+ }
+ return se;
+}
+
@Setup
public void setup() {
- executor.prepare();
+ this.se = compile();
}
@ProcessElement
public void processElement(ProcessContext c) {
- Row inputRow = c.element();
- @Nullable
- List rawResultValues =
- executor.execute(inputRow, null,
BeamSqlExpressionEnvironments.forRow(inputRow, null));
-
- if (rawResultValues != null) {
-List castResultValues =
-IntStream.range(0, outputSchema.getFieldCount())
-.mapToObj(i -> castField(rawResultValues, i))
-.collect(toList());
-
c.output(Row.withSchema(outputSchema).addValues(castResultValues).build());
+ assert se != null;
+ try {
+se.evaluate(new Object[] {outputSchema, c, CONTEXT_INSTANCE});
+ } catch (InvocationTargetException e) {
+throw new RuntimeException(
+"CalcFn failed to evaluate: " + processElementBlock, e.getCause());
+ }
+}
+ }
+
+ private static final Map rawTypeMap =
+ ImmutableMap.builder()
+ .put(TypeName.BYTE, Byte.class)
+ .put(TypeName.INT16, Short.class)
+ .put(TypeName.INT32, Integer.class)
+ .put(TypeName.INT64, Long.class)
+ .put(TypeName.FLOAT, Float.class)
+ .put(TypeName.DOUBLE, Double.class)
+ .build();
+
+ private Expression castOutput(Expression value, FieldType toType) {
Review comment:
It is never used outside of this class, I think leaving it here makes this
code easier to understand. If we have a second use case for it we can move it
out to `CalciteUtils`.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179567)
Time Spent: 4h (was: 3h 50m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 4h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179566&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179566
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:51
Start Date: 28/Dec/18 20:51
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244405554
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -68,15 +121,90 @@ public Calc copy(RelTraitSet traitSet, RelNode input,
RexProgram program) {
BeamCalcRel.class.getSimpleName(),
pinput);
PCollection upstream = pinput.get(0);
+ Schema outputSchema = CalciteUtils.toSchema(getRowType());
+
+ final SqlConformance conformance = SqlConformanceEnum.MYSQL_5;
+ final JavaTypeFactory typeFactory =
+ new JavaTypeFactoryImpl(BeamRelDataTypeSystem.INSTANCE) {
+@Override
+public Type getJavaClass(RelDataType type) {
+ if (type instanceof BasicSqlType || type instanceof
IntervalSqlType) {
+if (type.getSqlTypeName() == SqlTypeName.FLOAT) {
+ return type.isNullable() ? Float.class : float.class;
+}
+ }
+ return super.getJavaClass(type);
+}
+ };
+ final BlockBuilder builder = new BlockBuilder();
+
+ final PhysType physType =
+ PhysTypeImpl.of(typeFactory, getRowType(), JavaRowFormat.ARRAY,
false);
+
+ Expression input =
+ Expressions.convert_(Expressions.call(processContextParam,
"element"), Row.class);
+
+ final RexBuilder rexBuilder = getCluster().getRexBuilder();
+ final RelMetadataQuery mq = RelMetadataQuery.instance();
+ final RelOptPredicateList predicates =
mq.getPulledUpPredicates(getInput());
+ final RexSimplify simplify = new RexSimplify(rexBuilder, predicates,
false, RexUtil.EXECUTOR);
+ final RexProgram program =
BeamCalcRel.this.program.normalize(rexBuilder, simplify);
- BeamSqlExpressionExecutor executor = new
BeamSqlFnExecutor(BeamCalcRel.this.getProgram());
+ Expression condition =
+ RexToLixTranslator.translateCondition(
+ program,
+ typeFactory,
+ builder,
+ new InputGetterImpl(input, upstream.getSchema()),
+ null,
+ conformance);
- Schema schema = CalciteUtils.toSchema(rowType);
- PCollection projectStream =
- upstream
- .apply(ParDo.of(new CalcFn(executor,
CalciteUtils.toSchema(rowType
- .setRowSchema(schema);
- projectStream.setRowSchema(CalciteUtils.toSchema(getRowType()));
+ List expressions =
+ RexToLixTranslator.translateProjects(
+ program,
+ typeFactory,
+ conformance,
+ builder,
+ physType,
+ DataContext.ROOT,
+ new InputGetterImpl(input, upstream.getSchema()),
+ null);
+
+ // output = Row.withSchema(outputSchema)
+ Expression output = Expressions.call(Row.class, "withSchema",
outputSchemaParam);
+ Method addValue = Types.lookupMethod(Row.Builder.class, "addValue",
Object.class);
+
+ for (int index = 0; index < expressions.size(); index++) {
+Expression value = expressions.get(index);
+FieldType toType = outputSchema.getField(index).getType();
+
+// .addValue(value)
+output = Expressions.call(output, addValue, castOutput(value, toType));
+ }
+
+ // .build();
+ output = Expressions.call(output, "build");
+
+ // if (condition) {
+ // c.output(output);
+ // }
+ builder.add(
Review comment:
An example of the generated code is in the comment immediately above this
block.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179566)
Time Spent: 3h 50m (was: 3h 40m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
>
[jira] [Work logged] (BEAM-6319) BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
[
https://issues.apache.org/jira/browse/BEAM-6319?focusedWorklogId=179565&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179565
]
ASF GitHub Bot logged work on BEAM-6319:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:51
Start Date: 28/Dec/18 20:51
Worklog Time Spent: 10m
Work Description: boyuanzz commented on issue #7368:
[BEAM-6319][BEAM-6312][BEAM-6311] Disable BigQueryIo validation in
BigQueryToTableIT
URL: https://github.com/apache/beam/pull/7368#issuecomment-450425198
> According to the BigQuery docs, [table creation is eventually
consistent.](https://cloud.google.com/bigquery/troubleshooting-errors#metadata-errors-for-streaming-inserts)
It seems like this fix is just moving the race from pipeline creation to
worker setup. Does that code do something different that would allow it to
retry on this failure?
The retry strategy is only applied to querying. Other operation is one time
only. I'll handle the exception in next revision.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179565)
Time Spent: 20m (was: 10m)
> BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
> ---
>
> Key: BEAM-6319
> URL: https://issues.apache.org/jira/browse/BEAM-6319
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Reporter: Andrew Pilloud
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 20m
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PostCommit_Java/2213/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryToTableIT/testNewTypesQueryWithoutReshuffleWithCustom/
> {code}
> java.lang.IllegalArgumentException: Validation of query "SELECT bytes, date,
> time FROM [apache-beam-testing:bq_query_to_table_1545892541359_421.types]"
> failed. If the query depends on an earlier stage of the pipeline, This
> validation can be disabled using #withoutValidation.
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$TypedRead.validate(BigQueryIO.java:705)
> at
> org.apache.beam.sdk.Pipeline$ValidateVisitor.enterCompositeTransform(Pipeline.java:641)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:645)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:311)
> at
> org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:245)
> at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
> at org.apache.beam.sdk.Pipeline.validate(Pipeline.java:577)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:312)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.runBigQueryToTablePipeline(BigQueryToTableIT.java:111)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom(BigQueryToTableIT.java:307)
> 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.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.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:349)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179564&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179564
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:50
Start Date: 28/Dec/18 20:50
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244405355
##
File path:
sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCalcRel.java
##
@@ -68,15 +121,90 @@ public Calc copy(RelTraitSet traitSet, RelNode input,
RexProgram program) {
BeamCalcRel.class.getSimpleName(),
pinput);
PCollection upstream = pinput.get(0);
+ Schema outputSchema = CalciteUtils.toSchema(getRowType());
+
+ final SqlConformance conformance = SqlConformanceEnum.MYSQL_5;
+ final JavaTypeFactory typeFactory =
+ new JavaTypeFactoryImpl(BeamRelDataTypeSystem.INSTANCE) {
+@Override
+public Type getJavaClass(RelDataType type) {
+ if (type instanceof BasicSqlType || type instanceof
IntervalSqlType) {
+if (type.getSqlTypeName() == SqlTypeName.FLOAT) {
+ return type.isNullable() ? Float.class : float.class;
+}
+ }
+ return super.getJavaClass(type);
+}
+ };
+ final BlockBuilder builder = new BlockBuilder();
+
+ final PhysType physType =
+ PhysTypeImpl.of(typeFactory, getRowType(), JavaRowFormat.ARRAY,
false);
+
+ Expression input =
+ Expressions.convert_(Expressions.call(processContextParam,
"element"), Row.class);
+
+ final RexBuilder rexBuilder = getCluster().getRexBuilder();
+ final RelMetadataQuery mq = RelMetadataQuery.instance();
+ final RelOptPredicateList predicates =
mq.getPulledUpPredicates(getInput());
+ final RexSimplify simplify = new RexSimplify(rexBuilder, predicates,
false, RexUtil.EXECUTOR);
+ final RexProgram program =
BeamCalcRel.this.program.normalize(rexBuilder, simplify);
- BeamSqlExpressionExecutor executor = new
BeamSqlFnExecutor(BeamCalcRel.this.getProgram());
+ Expression condition =
+ RexToLixTranslator.translateCondition(
Review comment:
It should be always safe to call. It validates the program has a condition
and if not emits 'true'. The code generator will optimize away 'if (true)'
before generating the java code.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179564)
Time Spent: 3h 40m (was: 3.5h)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 3h 40m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179563&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179563
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:46
Start Date: 28/Dec/18 20:46
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244404682
##
File path:
sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlCastTest.java
##
@@ -67,35 +68,7 @@ public void testCastToDate() {
PAssert.that(result)
.containsInAnyOrder(
-Row.withSchema(resultType).addValues(1, new DateTime(2018, 10, 18,
0, 0)).build());
-
-pipeline.run();
- }
-
- @Test
- public void testCastToDate2() {
Review comment:
20181018 means year 20181018. This results in a overflow.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179563)
Time Spent: 3.5h (was: 3h 20m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 3.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5112) Investigate if Calcite can generate functions that we need
[
https://issues.apache.org/jira/browse/BEAM-5112?focusedWorklogId=179562&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179562
]
ASF GitHub Bot logged work on BEAM-5112:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:45
Start Date: 28/Dec/18 20:45
Worklog Time Spent: 10m
Work Description: apilloud commented on pull request #6417: [BEAM-5112]
Use Calcite codegen to implement BeamCalcRel
URL: https://github.com/apache/beam/pull/6417#discussion_r244404540
##
File path:
sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslSqlStdOperatorsTest.java
##
@@ -1047,7 +1046,9 @@ public void testTruncate() {
@SqlOperatorTest(name = "RAND", kind = "OTHER_FUNCTION")
public void testRand() {
ExpressionChecker checker =
-new ExpressionChecker().addExpr("RAND(c_integer)", new
Random(INTEGER_VALUE).nextDouble());
+new ExpressionChecker()
+.addExpr(
+"RAND(c_integer)", new Random(INTEGER_VALUE ^ (INTEGER_VALUE
<< 16)).nextDouble());
Review comment:
This is actually testing Calcite now. I'm guessing this behavior is
determined by the SQL standard, but we probably don't do the right thing if
there is more than one thread/worker.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179562)
Time Spent: 3h 20m (was: 3h 10m)
> Investigate if Calcite can generate functions that we need
> --
>
> Key: BEAM-5112
> URL: https://issues.apache.org/jira/browse/BEAM-5112
> Project: Beam
> Issue Type: Sub-task
> Components: dsl-sql
>Reporter: Rui Wang
>Assignee: Andrew Pilloud
>Priority: Major
> Time Spent: 3h 20m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179561&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179561
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:39
Start Date: 28/Dec/18 20:39
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244403631
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
+ FullWindowedValueCoder windowedValueCoder =
(FullWindowedValueCoder) javaCoder;
+ Coder windowCoder = windowedValueCoder.getWindowCoder();
+ if (IntervalWindowCoder.class.isInstance(windowCoder)) {
+fakeWindowingStrategyId = "generatedFixedWindowingStrategy" +
idGenerator.getId();
+try {
+ // Since the coder is the only needed from a WindowingStrategy,
the size field of one
Review comment:
Got it. Thanks~
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179561)
Time Spent: 3h 40m (was: 3.5h)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 3h 40m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179560&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179560
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:35
Start Date: 28/Dec/18 20:35
Worklog Time Spent: 10m
Work Description: mxm commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r244402840
##
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java
##
@@ -626,13 +650,17 @@ public void testSourceWithNoReaderDoesNotShutdown()
throws Exception {
try {
thread.start();
-List>>
localReaders =
-sourceWrapper.getLocalReaders();
-while (localReaders != null && !localReaders.isEmpty()) {
- Thread.sleep(200);
- // should stay alive
- assertThat(thread.isAlive(), is(true));
+Thread.sleep(200);
Review comment:
Comment added.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179560)
Time Spent: 1h 40m (was: 1.5h)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 1h 40m
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6294) Use Flink's redistribute for reshuffle.
[ https://issues.apache.org/jira/browse/BEAM-6294?focusedWorklogId=179559&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179559 ] ASF GitHub Bot logged work on BEAM-6294: Author: ASF GitHub Bot Created on: 28/Dec/18 20:31 Start Date: 28/Dec/18 20:31 Worklog Time Spent: 10m Work Description: robertwb commented on pull request #7360: [BEAM-6294] Use Flink rebalance for shuffle. URL: https://github.com/apache/beam/pull/7360 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179559) Time Spent: 50m (was: 40m) > Use Flink's redistribute for reshuffle. > --- > > Key: BEAM-6294 > URL: https://issues.apache.org/jira/browse/BEAM-6294 > Project: Beam > Issue Type: New Feature > Components: runner-flink, sdk-py-core >Reporter: Robert Bradshaw >Assignee: Robert Bradshaw >Priority: Major > Time Spent: 50m > Remaining Estimate: 0h > > Python needs to publish the URN over the FnAPI which is pretty easy, but > Flink also needs to ensure that the composite structure does not get fused. > Unlike with GBK, we can't assume all runners implement this as a primitive. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179558&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179558
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:29
Start Date: 28/Dec/18 20:29
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244401964
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
+ FullWindowedValueCoder windowedValueCoder =
(FullWindowedValueCoder) javaCoder;
+ Coder windowCoder = windowedValueCoder.getWindowCoder();
+ if (IntervalWindowCoder.class.isInstance(windowCoder)) {
+fakeWindowingStrategyId = "generatedFixedWindowingStrategy" +
idGenerator.getId();
+try {
+ // Since the coder is the only needed from a WindowingStrategy,
the size field of one
Review comment:
I meant creating a [WindowingStrategy
proto](https://github.com/apache/beam/blob/release-2.9.0/model/pipeline/src/main/proto/beam_runner_api.proto#598)
with all the fields but
[window_coder_id](https://github.com/apache/beam/blob/release-2.9.0/model/pipeline/src/main/proto/beam_runner_api.proto#L611)
unset (or set to something fake).
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179558)
Time Spent: 3.5h (was: 3h 20m)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 3.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179557&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179557
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:28
Start Date: 28/Dec/18 20:28
Worklog Time Spent: 10m
Work Description: mxm commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r244401699
##
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java
##
@@ -626,13 +650,17 @@ public void testSourceWithNoReaderDoesNotShutdown()
throws Exception {
try {
thread.start();
-List>>
localReaders =
-sourceWrapper.getLocalReaders();
-while (localReaders != null && !localReaders.isEmpty()) {
- Thread.sleep(200);
- // should stay alive
- assertThat(thread.isAlive(), is(true));
+Thread.sleep(200);
Review comment:
The test results are deterministic, this test cannot be flaky.
Will add the comment.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179557)
Time Spent: 1.5h (was: 1h 20m)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 1.5h
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5467) Python Flink ValidatesRunner job issues
[ https://issues.apache.org/jira/browse/BEAM-5467?focusedWorklogId=179554&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179554 ] ASF GitHub Bot logged work on BEAM-5467: Author: ASF GitHub Bot Created on: 28/Dec/18 20:24 Start Date: 28/Dec/18 20:24 Worklog Time Spent: 10m Work Description: mxm commented on issue #7376: [BEAM-5467] Increase test timeout for portable ValidatesRunner tests URL: https://github.com/apache/beam/pull/7376#issuecomment-450421581 Thanks for the quick review. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179554) Time Spent: 18h (was: 17h 50m) > Python Flink ValidatesRunner job issues > --- > > Key: BEAM-5467 > URL: https://issues.apache.org/jira/browse/BEAM-5467 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Critical > Labels: flake, portability-flink > Time Spent: 18h > Remaining Estimate: 0h > > > The job is currently flaky and does not provide a useful signal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5467) Python Flink ValidatesRunner job issues
[ https://issues.apache.org/jira/browse/BEAM-5467?focusedWorklogId=179555&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179555 ] ASF GitHub Bot logged work on BEAM-5467: Author: ASF GitHub Bot Created on: 28/Dec/18 20:24 Start Date: 28/Dec/18 20:24 Worklog Time Spent: 10m Work Description: mxm commented on pull request #7376: [BEAM-5467] Increase test timeout for portable ValidatesRunner tests URL: https://github.com/apache/beam/pull/7376 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179555) Time Spent: 18h 10m (was: 18h) > Python Flink ValidatesRunner job issues > --- > > Key: BEAM-5467 > URL: https://issues.apache.org/jira/browse/BEAM-5467 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Critical > Labels: flake, portability-flink > Time Spent: 18h 10m > Remaining Estimate: 0h > > > The job is currently flaky and does not provide a useful signal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6316) beam-sdks-python:portableWordCountBatch flaky: could not load docker image
[
https://issues.apache.org/jira/browse/BEAM-6316?focusedWorklogId=179553&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179553
]
ASF GitHub Bot logged work on BEAM-6316:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:23
Start Date: 28/Dec/18 20:23
Worklog Time Spent: 10m
Work Description: mxm commented on pull request #7377: [BEAM-6316] Fix
container image name for PreCommit PortableWordCount
URL: https://github.com/apache/beam/pull/7377
The image name changed after the multi-version build layout had been
merged. Until we allow passing a URL, we use the hardcoded name from the
Python
SDK.
Post-Commit Tests Status (on master branch)
Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark
--- | --- | --- | --- | --- | --- | --- | ---
Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/)
| --- | --- | --- | --- | --- | ---
Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)
[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)
Python | [](https://builds.apache.org/job/beam_PostCommit_Python_Verify/lastCompletedBuild/)
| --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)
[](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/)
| [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/)
| --- | --- | ---
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179553)
Time Spent: 10m
Remaining Estimate: 0h
> beam-sdks-python:portableWordCountBatch flaky: could not load docker image
> --
>
> Key: BEAM-6316
> URL: https://issues.apache.org/jira/browse/BEAM-6316
> Project: Beam
> Issue Type: Bug
> Components: runner-flink, test-failures
>Reporter: Andrew Pilloud
>Assignee: Maximilian Michels
>Priority: Major
> Time Spent: 10m
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/105/
> {code}
> 10:08:30 INFO:root:Using latest locally built Python SDK docker image.
> 10:08:31 Unable to find image
> 'jenkins-docker-apache.bintray.io/beam/flink-job-ser
[jira] [Created] (BEAM-6328) Provide environment option for JobServer container URL
Maximilian Michels created BEAM-6328: Summary: Provide environment option for JobServer container URL Key: BEAM-6328 URL: https://issues.apache.org/jira/browse/BEAM-6328 Project: Beam Issue Type: Improvement Components: sdk-py-core Reporter: Maximilian Michels If no endpoint is provided, the PortableRunner uses a hardcoded Docker image to start a JobServer. Similar to the SDK Docker image URL, we should allow specifying a URL for the JobServer container image. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179550&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179550
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 20:12
Start Date: 28/Dec/18 20:12
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244399099
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
+ FullWindowedValueCoder windowedValueCoder =
(FullWindowedValueCoder) javaCoder;
+ Coder windowCoder = windowedValueCoder.getWindowCoder();
+ if (IntervalWindowCoder.class.isInstance(windowCoder)) {
+fakeWindowingStrategyId = "generatedFixedWindowingStrategy" +
idGenerator.getId();
+try {
+ // Since the coder is the only needed from a WindowingStrategy,
the size field of one
Review comment:
Could you please explain more about how we can manually set the coder_id to
the right thing?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179550)
Time Spent: 3h 20m (was: 3h 10m)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 3h 20m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179549&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179549
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:58
Start Date: 28/Dec/18 19:58
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244396588
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
+ FullWindowedValueCoder windowedValueCoder =
(FullWindowedValueCoder) javaCoder;
+ Coder windowCoder = windowedValueCoder.getWindowCoder();
+ if (IntervalWindowCoder.class.isInstance(windowCoder)) {
+fakeWindowingStrategyId = "generatedFixedWindowingStrategy" +
idGenerator.getId();
+try {
+ // Since the coder is the only needed from a WindowingStrategy,
the size field of one
Review comment:
Yes, I agree. I'm suggesting we create a very fake one, manually setting the
coder_id to the right thing (regardless of whether it's understood by Java),
rather than a FixedWindows.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179549)
Time Spent: 3h 10m (was: 3h)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 3h 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179548&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179548
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:53
Start Date: 28/Dec/18 19:53
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244395596
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
+ FullWindowedValueCoder windowedValueCoder =
(FullWindowedValueCoder) javaCoder;
+ Coder windowCoder = windowedValueCoder.getWindowCoder();
+ if (IntervalWindowCoder.class.isInstance(windowCoder)) {
+fakeWindowingStrategyId = "generatedFixedWindowingStrategy" +
idGenerator.getId();
+try {
+ // Since the coder is the only needed from a WindowingStrategy,
the size field of one
Review comment:
Ideally, the WindowingStrategy should be instantiated in the SDK and passed
with pipeline proto to runner harness. However, dataflow runner harness doesn't
have pipeline proto thus dataflow generates this fake one in order to make it
work with fnapi.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179548)
Time Spent: 3h (was: 2h 50m)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 3h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179547&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179547
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:47
Start Date: 28/Dec/18 19:47
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244394426
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
+ FullWindowedValueCoder windowedValueCoder =
(FullWindowedValueCoder) javaCoder;
+ Coder windowCoder = windowedValueCoder.getWindowCoder();
+ if (IntervalWindowCoder.class.isInstance(windowCoder)) {
+fakeWindowingStrategyId = "generatedFixedWindowingStrategy" +
idGenerator.getId();
+try {
+ // Since the coder is the only needed from a WindowingStrategy,
the size field of one
+ // FixedWindows is meaningless here.
+ RunnerApi.MessageWithComponents windowingStrategyProto =
+ WindowingStrategyTranslation.toMessageProto(
+
WindowingStrategy.of(FixedWindows.of(Duration.standardSeconds(1))),
+ sdkComponents);
+ componentsBuilder.putWindowingStrategies(
+ fakeWindowingStrategyId,
windowingStrategyProto.getWindowingStrategy());
+
componentsBuilder.putAllCoders(windowingStrategyProto.getComponents().getCodersMap());
+ componentsBuilder.putAllEnvironments(
Review comment:
I didn't test these against non-java but I guess there is a huge possibility
to break non-java. This is another concern, I may not be able to extract window
coder and element coder from a non-java coder.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179547)
Time Spent: 2h 50m (was: 2h 40m)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 2h 50m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179545&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179545
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:44
Start Date: 28/Dec/18 19:44
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244393813
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
Review comment:
In that case, let's error if not. (Or otherwise discover the exception and
properly handle it.)
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179545)
Time Spent: 2.5h (was: 2h 20m)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 2.5h
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179546&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179546
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:44
Start Date: 28/Dec/18 19:44
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244393895
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
Review comment:
I'd rather not selectively lie, unless there's a good reason to do so.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179546)
Time Spent: 2h 40m (was: 2.5h)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 2h 40m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179544&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179544
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:40
Start Date: 28/Dec/18 19:40
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244393114
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
Review comment:
I'm not sure whether I understand the coder usage correctly, but feel like,
only the PCollection which is across the sdk-runner boundary cares the window
coder. Thus, it's ok just assign the GlobalWindow to these non-input-output
PCollections.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179544)
Time Spent: 2h 20m (was: 2h 10m)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 2h 20m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (BEAM-6327) Don't attempt to fuse subtransforms of primitive/known transforms.
Robert Bradshaw created BEAM-6327: - Summary: Don't attempt to fuse subtransforms of primitive/known transforms. Key: BEAM-6327 URL: https://issues.apache.org/jira/browse/BEAM-6327 Project: Beam Issue Type: New Feature Components: runner-direct Reporter: Robert Bradshaw Assignee: Daniel Oliveira Currently we must remove all sub-components of any known transform that may have an optional substructure, e.g. [https://github.com/apache/beam/blob/release-2.9.0/sdks/python/apache_beam/runners/portability/portable_runner.py#L126] (for GBK) and [https://github.com/apache/beam/pull/7360] (Reshuffle). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-6284) [FLAKE][beam_PostCommit_Java_ValidatesRunner_Dataflow] TestRunner fails with result UNKNOWN on succeeded job and checks passed
[ https://issues.apache.org/jira/browse/BEAM-6284?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730446#comment-16730446 ] Andrew Pilloud commented on BEAM-6284: -- https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_PortabilityApi_Dataflow/396/testReport/junit/org.apache.beam.sdk.transforms/CombineTest$BasicTests/testHotKeyCombiningWithAccumulationMode/ > [FLAKE][beam_PostCommit_Java_ValidatesRunner_Dataflow] TestRunner fails with > result UNKNOWN on succeeded job and checks passed > -- > > Key: BEAM-6284 > URL: https://issues.apache.org/jira/browse/BEAM-6284 > Project: Beam > Issue Type: Bug > Components: test-failures, testing >Reporter: Mikhail Gryzykhin >Assignee: Jason Kuster >Priority: Major > Labels: currently-failing > > _Use this form to file an issue for test failure:_ > * > https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/testReport/junit/org.apache.beam.sdk.transforms/ViewTest/testWindowedSideInputFixedToGlobal/ > Initial investigation: > According to logs all test-relevant checks have passed and it seem to be > testing framework failure. > > _After you've filled out the above details, please [assign the issue to an > individual|https://beam.apache.org/contribute/postcommits-guides/index.html#find_specialist]. > Assignee should [treat test failures as > high-priority|https://beam.apache.org/contribute/postcommits-policies/#assigned-failing-test], > helping to fix the issue or find a more appropriate owner. See [Apache Beam > Post-Commit > Policies|https://beam.apache.org/contribute/postcommits-policies]._ -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5955) [Flake][Java PortabilityAPI PostCommit] Flake due to testE2EBigtableWrite getting stuck.
[
https://issues.apache.org/jira/browse/BEAM-5955?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730451#comment-16730451
]
Andrew Pilloud commented on BEAM-5955:
--
https://builds.apache.org/job/beam_PostCommit_Java_PortabilityApi/539/
> [Flake][Java PortabilityAPI PostCommit] Flake due to testE2EBigtableWrite
> getting stuck.
>
>
> Key: BEAM-5955
> URL: https://issues.apache.org/jira/browse/BEAM-5955
> Project: Beam
> Issue Type: Bug
> Components: test-failures
>Reporter: Daniel Oliveira
>Priority: Minor
>
> [https://builds.apache.org/job/beam_PostCommit_Java_PortabilityApi_GradleBuild/36/]
> {noformat}
> SEVERE: 2018-11-02T18:12:37.831Z: Workflow failed. Causes: The Dataflow job
> appears to be stuck because no worker activity has been seen in the last 1h.
> You can get help with Cloud Dataflow at
> https://cloud.google.com/dataflow/support.
> {noformat}
> The test took 1h4m to run. TestE2EBigtableWrite usually takes about 6 min. to
> complete, based on Gradle build scans of successful runs, so the job seems to
> actually be getting stuck.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6231) Triage test failures introduced by use_executable_stage_bundle_execution
[
https://issues.apache.org/jira/browse/BEAM-6231?focusedWorklogId=179543&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179543
]
ASF GitHub Bot logged work on BEAM-6231:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:30
Start Date: 28/Dec/18 19:30
Worklog Time Spent: 10m
Work Description: boyuanzz commented on pull request #7356: [BEAM-6231]
Make Dataflow runner harness work with FixedWindow
URL: https://github.com/apache/beam/pull/7356#discussion_r244391412
##
File path:
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/graph/CreateExecutableStageNodeFunction.java
##
@@ -191,6 +195,37 @@ public Node apply(MutableNetwork input) {
Iterables.filter(input.nodes(), InstructionOutputNode.class)) {
InstructionOutput instructionOutput = node.getInstructionOutput();
+ // If this is the input PCollection or the output PCollection for an
ExecutableStage, it's
+ // necessary to check whether the window coder is not a GlobalWindow
coder.
+ if (isExecutableStageInputPCollection(input, node)
+ || isExecutableStageOutputPCollection(input, node)) {
+Coder javaCoder =
+
CloudObjects.coderFromCloudObject(CloudObject.fromSpec(instructionOutput.getCodec()));
+// For now, Dataflow runner harness only deal with FixedWindow.
+if (FullWindowedValueCoder.class.isInstance(javaCoder)) {
Review comment:
From what I know so far, if there is a window, then FullWindowedValueCoder
is the only coder. I feel like the coder I got from v1b3 should be a
FullWindowedValueCoder.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179543)
Time Spent: 2h 10m (was: 2h)
> Triage test failures introduced by use_executable_stage_bundle_execution
>
>
> Key: BEAM-6231
> URL: https://issues.apache.org/jira/browse/BEAM-6231
> Project: Beam
> Issue Type: Test
> Components: runner-dataflow
>Reporter: Boyuan Zhang
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 2h 10m
> Remaining Estimate: 0h
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6294) Use Flink's redistribute for reshuffle.
[
https://issues.apache.org/jira/browse/BEAM-6294?focusedWorklogId=179541&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179541
]
ASF GitHub Bot logged work on BEAM-6294:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:29
Start Date: 28/Dec/18 19:29
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #7360: [BEAM-6294]
Use Flink rebalance for shuffle.
URL: https://github.com/apache/beam/pull/7360#discussion_r244390200
##
File path:
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobInvocation.java
##
@@ -98,41 +99,71 @@ private FlinkJobInvocation(
private PipelineResult runPipeline() throws Exception {
MetricsEnvironment.setMetricsSupported(false);
+FlinkPortablePipelineTranslator translator;
+if (!pipelineOptions.isStreaming() && !hasUnboundedPCollections(pipeline))
{
+ // TODO: Do we need to inspect for unbounded sources before fusing?
+ translator = FlinkBatchPortablePipelineTranslator.createTranslator();
+} else {
+ translator = new FlinkStreamingPortablePipelineTranslator();
+}
+return runPipelineWithTranslator(translator);
+ }
+
+ private
+ PipelineResult
runPipelineWithTranslator(FlinkPortablePipelineTranslator translator)
+ throws Exception {
LOG.info("Translating pipeline to Flink program.");
+
+// Don't let the fuser fuse any subcomponents.
+RunnerApi.Pipeline trimmedPipeline = makeKnownUrnsPrimitives(pipeline,
translator.knownUrns());
+System.out.println(trimmedPipeline.getComponents().getTransformsMap());
+
// Fused pipeline proto.
-RunnerApi.Pipeline fusedPipeline =
GreedyPipelineFuser.fuse(pipeline).toPipeline();
+RunnerApi.Pipeline fusedPipeline =
GreedyPipelineFuser.fuse(trimmedPipeline).toPipeline();
JobInfo jobInfo =
JobInfo.create(
id,
pipelineOptions.getJobName(),
retrievalToken,
PipelineOptionsTranslation.toProto(pipelineOptions));
-final JobExecutionResult result;
-if (!pipelineOptions.isStreaming() &&
!hasUnboundedPCollections(fusedPipeline)) {
- // TODO: Do we need to inspect for unbounded sources before fusing?
- // batch translation
- FlinkBatchPortablePipelineTranslator translator =
- FlinkBatchPortablePipelineTranslator.createTranslator();
- FlinkBatchPortablePipelineTranslator.BatchTranslationContext context =
- FlinkBatchPortablePipelineTranslator.createTranslationContext(
- jobInfo, pipelineOptions, confDir, filesToStage);
- translator.translate(context, fusedPipeline);
- result =
context.getExecutionEnvironment().execute(pipelineOptions.getJobName());
-} else {
- // streaming translation
- FlinkStreamingPortablePipelineTranslator translator =
- new FlinkStreamingPortablePipelineTranslator();
- FlinkStreamingPortablePipelineTranslator.StreamingTranslationContext
context =
- FlinkStreamingPortablePipelineTranslator.createTranslationContext(
- jobInfo, pipelineOptions, confDir, filesToStage);
- translator.translate(context, fusedPipeline);
- result =
context.getExecutionEnvironment().execute(pipelineOptions.getJobName());
-}
+FlinkPortablePipelineTranslator.Executor executor =
+translator.translate(
+translator.createTranslationContext(jobInfo, pipelineOptions,
confDir, filesToStage),
+fusedPipeline);
+final JobExecutionResult result =
executor.execute(pipelineOptions.getJobName());
return FlinkRunner.createPipelineResult(result, pipelineOptions);
}
+ private RunnerApi.Pipeline makeKnownUrnsPrimitives(
+ RunnerApi.Pipeline pipeline, Set knownUrns) {
+RunnerApi.Pipeline.Builder trimmedPipeline = pipeline.toBuilder();
+for (String ptransformId :
pipeline.getComponents().getTransformsMap().keySet()) {
+ if (knownUrns.contains(
+
pipeline.getComponents().getTransformsOrThrow(ptransformId).getSpec().getUrn()))
{
+System.out.println("FOUND " + ptransformId);
Review comment:
Fixed here and elsewhere.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179541)
Time Spent: 0.5h (was: 20m)
> Use Flink's redistribute for reshuffle.
> ---
>
> Key: BEAM-6294
> URL: https://issues.apache.org/jira/browse/BEAM-6294
> Project: Beam
> Issue Type: New Feature
> Components: ru
[jira] [Work logged] (BEAM-6294) Use Flink's redistribute for reshuffle.
[
https://issues.apache.org/jira/browse/BEAM-6294?focusedWorklogId=179542&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179542
]
ASF GitHub Bot logged work on BEAM-6294:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:29
Start Date: 28/Dec/18 19:29
Worklog Time Spent: 10m
Work Description: robertwb commented on pull request #7360: [BEAM-6294]
Use Flink rebalance for shuffle.
URL: https://github.com/apache/beam/pull/7360#discussion_r244391280
##
File path:
runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkJobInvocation.java
##
@@ -98,41 +99,71 @@ private FlinkJobInvocation(
private PipelineResult runPipeline() throws Exception {
MetricsEnvironment.setMetricsSupported(false);
+FlinkPortablePipelineTranslator translator;
+if (!pipelineOptions.isStreaming() && !hasUnboundedPCollections(pipeline))
{
+ // TODO: Do we need to inspect for unbounded sources before fusing?
+ translator = FlinkBatchPortablePipelineTranslator.createTranslator();
+} else {
+ translator = new FlinkStreamingPortablePipelineTranslator();
+}
+return runPipelineWithTranslator(translator);
+ }
+
+ private
+ PipelineResult
runPipelineWithTranslator(FlinkPortablePipelineTranslator translator)
+ throws Exception {
LOG.info("Translating pipeline to Flink program.");
+
+// Don't let the fuser fuse any subcomponents.
+RunnerApi.Pipeline trimmedPipeline = makeKnownUrnsPrimitives(pipeline,
translator.knownUrns());
+System.out.println(trimmedPipeline.getComponents().getTransformsMap());
+
// Fused pipeline proto.
-RunnerApi.Pipeline fusedPipeline =
GreedyPipelineFuser.fuse(pipeline).toPipeline();
+RunnerApi.Pipeline fusedPipeline =
GreedyPipelineFuser.fuse(trimmedPipeline).toPipeline();
JobInfo jobInfo =
JobInfo.create(
id,
pipelineOptions.getJobName(),
retrievalToken,
PipelineOptionsTranslation.toProto(pipelineOptions));
-final JobExecutionResult result;
-if (!pipelineOptions.isStreaming() &&
!hasUnboundedPCollections(fusedPipeline)) {
- // TODO: Do we need to inspect for unbounded sources before fusing?
- // batch translation
- FlinkBatchPortablePipelineTranslator translator =
- FlinkBatchPortablePipelineTranslator.createTranslator();
- FlinkBatchPortablePipelineTranslator.BatchTranslationContext context =
- FlinkBatchPortablePipelineTranslator.createTranslationContext(
- jobInfo, pipelineOptions, confDir, filesToStage);
- translator.translate(context, fusedPipeline);
- result =
context.getExecutionEnvironment().execute(pipelineOptions.getJobName());
-} else {
- // streaming translation
- FlinkStreamingPortablePipelineTranslator translator =
- new FlinkStreamingPortablePipelineTranslator();
- FlinkStreamingPortablePipelineTranslator.StreamingTranslationContext
context =
- FlinkStreamingPortablePipelineTranslator.createTranslationContext(
- jobInfo, pipelineOptions, confDir, filesToStage);
- translator.translate(context, fusedPipeline);
- result =
context.getExecutionEnvironment().execute(pipelineOptions.getJobName());
-}
+FlinkPortablePipelineTranslator.Executor executor =
+translator.translate(
+translator.createTranslationContext(jobInfo, pipelineOptions,
confDir, filesToStage),
+fusedPipeline);
+final JobExecutionResult result =
executor.execute(pipelineOptions.getJobName());
return FlinkRunner.createPipelineResult(result, pipelineOptions);
}
+ private RunnerApi.Pipeline makeKnownUrnsPrimitives(
+ RunnerApi.Pipeline pipeline, Set knownUrns) {
+RunnerApi.Pipeline.Builder trimmedPipeline = pipeline.toBuilder();
+for (String ptransformId :
pipeline.getComponents().getTransformsMap().keySet()) {
+ if (knownUrns.contains(
+
pipeline.getComponents().getTransformsOrThrow(ptransformId).getSpec().getUrn()))
{
+System.out.println("FOUND " + ptransformId);
+removeDescendants(trimmedPipeline, ptransformId);
Review comment:
Yes, this would be nice. Non-trivial due to the way the Java fuser is
structured. Filed https://issues.apache.org/jira/browse/BEAM-6327
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179542)
Time Spent: 40m (was: 0.5h)
> Use Flink's redistribute for reshuffle.
> ---
>
> Key:
[jira] [Commented] (BEAM-6289) Running a join on two Cassandra tables using FlinkRunner fails
[
https://issues.apache.org/jira/browse/BEAM-6289?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730450#comment-16730450
]
Shahar Frank commented on BEAM-6289:
Great. Thanx.
I will give this a try on a remote Flink cluster I have at work on Sunday.
Have a great weekend.
> Running a join on two Cassandra tables using FlinkRunner fails
> --
>
> Key: BEAM-6289
> URL: https://issues.apache.org/jira/browse/BEAM-6289
> Project: Beam
> Issue Type: Bug
> Components: io-java-cassandra, runner-flink
>Affects Versions: 2.8.0, 2.9.0
> Environment: Tested on Ubuntu 18
> Beam 2.8
> Tested with Flink:
> 1) [local]
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
> Tested using Cassandra [cqlsh 5.0.1 | Cassandra 3.11.3 | CQL spec 3.4.4 |
> Native protocol v4]:
> 1) In a local container
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
>Reporter: Shahar Frank
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: FlinkRunner, beam, bug, cassandra, flink, join
> Attachments: direct_runner_build.log, flink_runner_build.log
>
>
> Can't make a simple join on two Cassandra tables when using FlinkRunner.
> The same code works with a DirectRunner fails when used with FlinkRunner
> giving these (as well as many other) errors:
> {code:java}
> Caused by: akka.pattern.AskTimeoutException: Ask timed out on
> [Actor[akka://flink/user/dispatchere1f5abe7-6299-43ea-9182-24a2193e078f#-1757043920]]
> after [1 ms]. Sender[null] sent message of type
> "org.apache.flink.runtime.rpc.messages.LocalFencedMessage".
> at
> akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
> at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
> at
> scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
> at
> akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
> at java.lang.Thread.run(Thread.java:748)
> {code}
>
> The code can be found [here|https://github.com/srfrnk/beam-playground]
> Steps to reproduce:
> # Clone the repo to a linux (I;m on Ubuntu 18 but any *nix system would
> probably work - i.e. repl.it)
> # Follow the README to set up a Cassandra container + schema
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=flink > output/build.log
> 2>&1{code}
> to use FlinkRunner. See error in log at ./output/build.log
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=direct > output/build.log
> 2>&1{code}
> to use DirectRunner. See error in log at ./output/build.log
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5415) [beam_PostCommit_Python_Verify] test is failing
[ https://issues.apache.org/jira/browse/BEAM-5415?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730448#comment-16730448 ] Andrew Pilloud commented on BEAM-5415: -- https://builds.apache.org/job/beam_PostCommit_Python_Verify/6939 > [beam_PostCommit_Python_Verify] test is failing > --- > > Key: BEAM-5415 > URL: https://issues.apache.org/jira/browse/BEAM-5415 > Project: Beam > Issue Type: Bug > Components: test-failures >Reporter: Batkhuyag Batsaikhan >Assignee: Valentyn Tymofieiev >Priority: Major > > Failing job url: > [https://builds.apache.org/job/beam_PostCommit_Python_Verify/6003/] > Most likely culprit PR: https://github.com/apache/beam/pull/6403 -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-6284) [FLAKE][beam_PostCommit_Java_ValidatesRunner_Dataflow] TestRunner fails with result UNKNOWN on succeeded job and checks passed
[ https://issues.apache.org/jira/browse/BEAM-6284?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730445#comment-16730445 ] Andrew Pilloud commented on BEAM-6284: -- https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_PortabilityApi_Dataflow/395/ > [FLAKE][beam_PostCommit_Java_ValidatesRunner_Dataflow] TestRunner fails with > result UNKNOWN on succeeded job and checks passed > -- > > Key: BEAM-6284 > URL: https://issues.apache.org/jira/browse/BEAM-6284 > Project: Beam > Issue Type: Bug > Components: test-failures, testing >Reporter: Mikhail Gryzykhin >Assignee: Jason Kuster >Priority: Major > Labels: currently-failing > > _Use this form to file an issue for test failure:_ > * > https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/testReport/junit/org.apache.beam.sdk.transforms/ViewTest/testWindowedSideInputFixedToGlobal/ > Initial investigation: > According to logs all test-relevant checks have passed and it seem to be > testing framework failure. > > _After you've filled out the above details, please [assign the issue to an > individual|https://beam.apache.org/contribute/postcommits-guides/index.html#find_specialist]. > Assignee should [treat test failures as > high-priority|https://beam.apache.org/contribute/postcommits-policies/#assigned-failing-test], > helping to fix the issue or find a more appropriate owner. See [Apache Beam > Post-Commit > Policies|https://beam.apache.org/contribute/postcommits-policies]._ -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179539&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179539
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:15
Start Date: 28/Dec/18 19:15
Worklog Time Spent: 10m
Work Description: tweise commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r244388633
##
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java
##
@@ -626,13 +650,17 @@ public void testSourceWithNoReaderDoesNotShutdown()
throws Exception {
try {
thread.start();
-List>>
localReaders =
-sourceWrapper.getLocalReaders();
-while (localReaders != null && !localReaders.isEmpty()) {
- Thread.sleep(200);
- // should stay alive
- assertThat(thread.isAlive(), is(true));
+Thread.sleep(200);
Review comment:
This looks like a recipe for non deterministic test results.. But I don't
have any better idea either. Would suggest to add above as comment to the code
at least.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179539)
Time Spent: 1h 20m (was: 1h 10m)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 1h 20m
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-6289) Running a join on two Cassandra tables using FlinkRunner fails
[
https://issues.apache.org/jira/browse/BEAM-6289?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730441#comment-16730441
]
Maximilian Michels commented on BEAM-6289:
--
{quote} The issue was first detected on a remote Flink cluster - I will recheck
it with this demo code - could this then be some other issue then?{quote}
Thanks. Keep in mind that you will still see the error message but the job
should start to run anyways on the remote cluster. Just the client connection
fails with the timeout being set to 10 seconds.
{quote}
And can you please tell me what the akka.ask.timeout value which is currently
being used? Is it very short? Because this entire code fail within a few
seconds - well below 10 seconds - so it's not that long - is it?
{quote}
It is 10 seconds. In my test setup, the job submission process exceeds 10
seconds.
> Running a join on two Cassandra tables using FlinkRunner fails
> --
>
> Key: BEAM-6289
> URL: https://issues.apache.org/jira/browse/BEAM-6289
> Project: Beam
> Issue Type: Bug
> Components: io-java-cassandra, runner-flink
>Affects Versions: 2.8.0, 2.9.0
> Environment: Tested on Ubuntu 18
> Beam 2.8
> Tested with Flink:
> 1) [local]
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
> Tested using Cassandra [cqlsh 5.0.1 | Cassandra 3.11.3 | CQL spec 3.4.4 |
> Native protocol v4]:
> 1) In a local container
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
>Reporter: Shahar Frank
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: FlinkRunner, beam, bug, cassandra, flink, join
> Attachments: direct_runner_build.log, flink_runner_build.log
>
>
> Can't make a simple join on two Cassandra tables when using FlinkRunner.
> The same code works with a DirectRunner fails when used with FlinkRunner
> giving these (as well as many other) errors:
> {code:java}
> Caused by: akka.pattern.AskTimeoutException: Ask timed out on
> [Actor[akka://flink/user/dispatchere1f5abe7-6299-43ea-9182-24a2193e078f#-1757043920]]
> after [1 ms]. Sender[null] sent message of type
> "org.apache.flink.runtime.rpc.messages.LocalFencedMessage".
> at
> akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
> at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
> at
> scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
> at
> akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
> at java.lang.Thread.run(Thread.java:748)
> {code}
>
> The code can be found [here|https://github.com/srfrnk/beam-playground]
> Steps to reproduce:
> # Clone the repo to a linux (I;m on Ubuntu 18 but any *nix system would
> probably work - i.e. repl.it)
> # Follow the README to set up a Cassandra container + schema
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=flink > output/build.log
> 2>&1{code}
> to use FlinkRunner. See error in log at ./output/build.log
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=direct > output/build.log
> 2>&1{code}
> to use DirectRunner. See error in log at ./output/build.log
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5467) Python Flink ValidatesRunner job issues
[ https://issues.apache.org/jira/browse/BEAM-5467?focusedWorklogId=179538&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179538 ] ASF GitHub Bot logged work on BEAM-5467: Author: ASF GitHub Bot Created on: 28/Dec/18 19:11 Start Date: 28/Dec/18 19:11 Worklog Time Spent: 10m Work Description: mxm commented on issue #7376: [BEAM-5467] Increase test timeout for portable ValidatesRunner tests URL: https://github.com/apache/beam/pull/7376#issuecomment-450411814 Run Python PreCommit This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179538) Time Spent: 17h 50m (was: 17h 40m) > Python Flink ValidatesRunner job issues > --- > > Key: BEAM-5467 > URL: https://issues.apache.org/jira/browse/BEAM-5467 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Critical > Labels: flake, portability-flink > Time Spent: 17h 50m > Remaining Estimate: 0h > > > The job is currently flaky and does not provide a useful signal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179537&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179537
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:10
Start Date: 28/Dec/18 19:10
Worklog Time Spent: 10m
Work Description: mxm commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r244387670
##
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java
##
@@ -626,13 +650,17 @@ public void testSourceWithNoReaderDoesNotShutdown()
throws Exception {
try {
thread.start();
-List>>
localReaders =
-sourceWrapper.getLocalReaders();
-while (localReaders != null && !localReaders.isEmpty()) {
- Thread.sleep(200);
- // should stay alive
- assertThat(thread.isAlive(), is(true));
+Thread.sleep(200);
Review comment:
It it just a test that the source does not shut down immediately. The only
way to verify is to wait for some amount of time. More importantly the
watermark progressing which follows should trigger shutdown immediately if
enabled which is verified with `assertThat(sourceWrapper.isRunning(),
is(true));`
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179537)
Time Spent: 1h 10m (was: 1h)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 1h 10m
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5396) Flink portable runner savepoint / upgrade support
[ https://issues.apache.org/jira/browse/BEAM-5396?focusedWorklogId=179536&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179536 ] ASF GitHub Bot logged work on BEAM-5396: Author: ASF GitHub Bot Created on: 28/Dec/18 19:06 Start Date: 28/Dec/18 19:06 Worklog Time Spent: 10m Work Description: mxm commented on issue #7362: [BEAM-5396] Assign portable operator uids URL: https://github.com/apache/beam/pull/7362#issuecomment-450411071 >There is a commit from another PR included here, is that intentional? Yes, this is based on #7359 (see description) because the test wouldn't otherwise run. > Also, we you able to reproduce the restore error? As you wrote in the JIRA, it depends on the pipeline. We shouldn't see the restore error anymore because of the uid assigned to all stateful operators. Before it would depend on the internal operator ids which only works if the pipeline generation is deterministic. Perhaps you could re-run your pipeline with this PR to verify that the restore works? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179536) Time Spent: 2h 50m (was: 2h 40m) > Flink portable runner savepoint / upgrade support > - > > Key: BEAM-5396 > URL: https://issues.apache.org/jira/browse/BEAM-5396 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Time Spent: 2h 50m > Remaining Estimate: 0h > > The portable Flink runner needs to support Flink savepoints for production > use. It should be possible to upgrade a stateful portable Beam pipeline that > runs on Flink, which involves taking a savepoint and then starting the new > version of the pipeline from that savepoint. The potential issues with > pipeline evolution and migration are similar to those when using the Flink > DataStream API (schema / name changes etc.). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5386) Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
[
https://issues.apache.org/jira/browse/BEAM-5386?focusedWorklogId=179535&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179535
]
ASF GitHub Bot logged work on BEAM-5386:
Author: ASF GitHub Bot
Created on: 28/Dec/18 19:03
Start Date: 28/Dec/18 19:03
Worklog Time Spent: 10m
Work Description: tweise commented on pull request #7349: [BEAM-5386]
Prevent CheckpointMarks from not getting acknowledged
URL: https://github.com/apache/beam/pull/7349#discussion_r244386367
##
File path:
runners/flink/src/test/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapperTest.java
##
@@ -626,13 +650,17 @@ public void testSourceWithNoReaderDoesNotShutdown()
throws Exception {
try {
thread.start();
-List>>
localReaders =
-sourceWrapper.getLocalReaders();
-while (localReaders != null && !localReaders.isEmpty()) {
- Thread.sleep(200);
- // should stay alive
- assertThat(thread.isAlive(), is(true));
+Thread.sleep(200);
Review comment:
How did you arrive at this sleep interval?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179535)
Time Spent: 1h (was: 50m)
> Flink Runner gets progressively stuck when Pubsub subscription is nearly empty
> --
>
> Key: BEAM-5386
> URL: https://issues.apache.org/jira/browse/BEAM-5386
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp, runner-flink
>Affects Versions: 2.6.0
>Reporter: Encho Mishinev
>Assignee: Chamikara Jayalath
>Priority: Major
> Time Spent: 1h
> Remaining Estimate: 0h
>
> I am running the Flink runner on Apache Beam 2.6.0.
> My pipeline involves reading from Google Cloud Pubsub. The problem is that
> whenever there are few messages left in the subscription I'm reading from,
> the whole job becomes progressively slower and slower, Flink's checkpoints
> start taking much more time and messages seem to not get properly
> acknowledged.
> This happens only whenever the subscription is nearly empty. For example when
> running 13 taskmanagers with parallelism of 52 for the job and a subscription
> that has 122 000 000 messages, you start feeling the slowing down after there
> are only 1 000 000 - 2 000 000 messages left.
> In one of my tests the job processed nearly 122 000 000 messages in an hour
> and then spent over 30 minutes attempting to do the few hundred thousand
> left. In the end it was reading a few hundred messages a minute and not
> reading at all for some periods. Upon stopping it the subscription still had
> 235 unacknowledged messages, even though Flink's element count was higher
> than the amount of messages I had loaded. The only explanation is that the
> messages did not get properly acknowledged and were resent.
> I have set up the subscriptions to a large acknowledgment deadline, but that
> does not help.
> I did smaller tests on subscriptions with 100 000 messages and a job that
> simply reads and does nothing else. The problem is still evident. With
> parallelism of 52 the job gets slow right away. Takes over 5min to read about
> 100 000 messages and a few hundred seem to keep cycling through never being
> acknowledged.
> On the other hand a parallelism of 1 works fine until there are about 5000
> messages left, and then slows down similarly.
> Parallelism of 16 reads about 75 000 of the 100 000 immediately (a few
> seconds) and then proceeds to slowly work on the other 25 000 for minutes.
> The PubsubIO connector is provided by Beam so I suspect the problem to be in
> Beam's Flink runner rather than Flink itself.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5396) Flink portable runner savepoint / upgrade support
[ https://issues.apache.org/jira/browse/BEAM-5396?focusedWorklogId=179534&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179534 ] ASF GitHub Bot logged work on BEAM-5396: Author: ASF GitHub Bot Created on: 28/Dec/18 18:57 Start Date: 28/Dec/18 18:57 Worklog Time Spent: 10m Work Description: tweise commented on issue #7362: [BEAM-5396] Assign portable operator uids URL: https://github.com/apache/beam/pull/7362#issuecomment-450409755 @mxm thank! There is a commit from another PR included here, is that intentional? Also, we you able to reproduce the restore error? If not, then I would need to do some more extensive testing for this. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179534) Time Spent: 2h 40m (was: 2.5h) > Flink portable runner savepoint / upgrade support > - > > Key: BEAM-5396 > URL: https://issues.apache.org/jira/browse/BEAM-5396 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Time Spent: 2h 40m > Remaining Estimate: 0h > > The portable Flink runner needs to support Flink savepoints for production > use. It should be possible to upgrade a stateful portable Beam pipeline that > runs on Flink, which involves taking a savepoint and then starting the new > version of the pipeline from that savepoint. The potential issues with > pipeline evolution and migration are similar to those when using the Flink > DataStream API (schema / name changes etc.). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5396) Flink portable runner savepoint / upgrade support
[ https://issues.apache.org/jira/browse/BEAM-5396?focusedWorklogId=179533&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179533 ] ASF GitHub Bot logged work on BEAM-5396: Author: ASF GitHub Bot Created on: 28/Dec/18 18:50 Start Date: 28/Dec/18 18:50 Worklog Time Spent: 10m Work Description: mxm commented on issue #7362: [BEAM-5396] Assign portable operator uids URL: https://github.com/apache/beam/pull/7362#issuecomment-450408705 Run Portable_Python PreCommit This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179533) Time Spent: 2.5h (was: 2h 20m) > Flink portable runner savepoint / upgrade support > - > > Key: BEAM-5396 > URL: https://issues.apache.org/jira/browse/BEAM-5396 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Time Spent: 2.5h > Remaining Estimate: 0h > > The portable Flink runner needs to support Flink savepoints for production > use. It should be possible to upgrade a stateful portable Beam pipeline that > runs on Flink, which involves taking a savepoint and then starting the new > version of the pipeline from that savepoint. The potential issues with > pipeline evolution and migration are similar to those when using the Flink > DataStream API (schema / name changes etc.). -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-4681) Integrate support for timers using the portability APIs into Flink
[ https://issues.apache.org/jira/browse/BEAM-4681?focusedWorklogId=179532&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179532 ] ASF GitHub Bot logged work on BEAM-4681: Author: ASF GitHub Bot Created on: 28/Dec/18 18:48 Start Date: 28/Dec/18 18:48 Worklog Time Spent: 10m Work Description: mxm commented on issue #7359: [BEAM-4681] Address synchronization issue for portable timers URL: https://github.com/apache/beam/pull/7359#issuecomment-450408419 No new test failures in `PortableValidatesRunner Streaming`. We need to address those separately: https://issues.apache.org/jira/browse/BEAM-6326 This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179532) Time Spent: 25h 40m (was: 25.5h) > Integrate support for timers using the portability APIs into Flink > -- > > Key: BEAM-4681 > URL: https://issues.apache.org/jira/browse/BEAM-4681 > Project: Beam > Issue Type: Sub-task > Components: runner-flink >Reporter: Luke Cwik >Assignee: Maximilian Michels >Priority: Major > Labels: portability, portability-flink > Fix For: 2.9.0 > > Time Spent: 25h 40m > Remaining Estimate: 0h > > Consider using the code produced in BEAM-4658 to support timers. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (BEAM-6326) Fix test failures in streaming mode of PortableValidatesRunner
Maximilian Michels created BEAM-6326:
Summary: Fix test failures in streaming mode of
PortableValidatesRunner
Key: BEAM-6326
URL: https://issues.apache.org/jira/browse/BEAM-6326
Project: Beam
Issue Type: Test
Components: runner-flink
Reporter: Maximilian Michels
As of BEAM-6009, the tests are run separately for batch and streaming. This has
revealed issues with a couple of tests which need to be addressed.
The Gradle task is: {{portableValidatesRunnerStreaming}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5467) Python Flink ValidatesRunner job issues
[ https://issues.apache.org/jira/browse/BEAM-5467?focusedWorklogId=179531&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179531 ] ASF GitHub Bot logged work on BEAM-5467: Author: ASF GitHub Bot Created on: 28/Dec/18 18:39 Start Date: 28/Dec/18 18:39 Worklog Time Spent: 10m Work Description: mxm commented on issue #7376: [BEAM-5467] Increase test timeout for portable ValidatesRunner tests URL: https://github.com/apache/beam/pull/7376#issuecomment-450407137 CC @tweise This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179531) Time Spent: 17h 40m (was: 17.5h) > Python Flink ValidatesRunner job issues > --- > > Key: BEAM-5467 > URL: https://issues.apache.org/jira/browse/BEAM-5467 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Critical > Labels: flake, portability-flink > Time Spent: 17h 40m > Remaining Estimate: 0h > > > The job is currently flaky and does not provide a useful signal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5467) Python Flink ValidatesRunner job issues
[ https://issues.apache.org/jira/browse/BEAM-5467?focusedWorklogId=179529&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179529 ] ASF GitHub Bot logged work on BEAM-5467: Author: ASF GitHub Bot Created on: 28/Dec/18 18:31 Start Date: 28/Dec/18 18:31 Worklog Time Spent: 10m Work Description: mxm commented on pull request #7376: [BEAM-5467] Increase test timeout for portable ValidatesRunner tests URL: https://github.com/apache/beam/pull/7376 The test `test_pardo_unfusable_side_inputs` runs two pipelines which each take about 17 seconds to run on Flink. The default timeout of 30 seconds is too low. Increasing to 60 seconds. Post-Commit Tests Status (on master branch) Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark --- | --- | --- | --- | --- | --- | --- | --- Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | --- | --- | --- | --- Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/) [](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/) Python | [](https://builds.apache.org/job/beam_PostCommit_Python_Verify/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/) [](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | --- | --- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179529) Time Spent: 17h 20m (was: 17h 10m) > Python Flink ValidatesRunner job issues > --- > > Key: BEAM-5467 > URL: https://issues.apache.org/jira/browse/BEAM-5467 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Critical > Labels: flake, portability-flink > Time Spent: 17h 20m > Remaining Estimate: 0h > > > The job is currently flaky and does not provide a useful signal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5936) [beam_PreCommit_Java_Cron] Flake due to flink.PortableStateExecutionTest
[ https://issues.apache.org/jira/browse/BEAM-5936?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Maximilian Michels resolved BEAM-5936. -- Resolution: Fixed Assignee: Maximilian Michels Fix Version/s: 2.10.0 I believe this should be fixed via BEAM-6283. Resolving but feel free to reopen if the problem persists. > [beam_PreCommit_Java_Cron] Flake due to flink.PortableStateExecutionTest > > > Key: BEAM-5936 > URL: https://issues.apache.org/jira/browse/BEAM-5936 > Project: Beam > Issue Type: Bug > Components: runner-flink, test-failures >Reporter: Daniel Oliveira >Assignee: Maximilian Michels >Priority: Minor > Fix For: 2.10.0 > > > [https://builds.apache.org/job/beam_PreCommit_Java_Cron/532/] > > This test failed due to > org.apache.beam.runners.flink.PortableStateExecutionTest.testExecution > failing. The next execution didn't fail with this issue so this is apparently > a flake. > Exact error was: > > java.lang.AssertionError: > Expected: is > but: was > > Please modify the priority of this bug if this error occurs again or if the > root cause is an existing bug, mark it somewhere. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-5467) Python Flink ValidatesRunner job issues
[
https://issues.apache.org/jira/browse/BEAM-5467?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730417#comment-16730417
]
Maximilian Michels commented on BEAM-5467:
--
The culprit is now {{test_pardo_unfusable_side_inputs}}. Not seeing any
failures of the {{test_assert_that}} anymore.
The issue is that the test contains two pipelines which each take about 17
seconds to run on Flink. The default timeout of 30 seconds is too low.
> Python Flink ValidatesRunner job issues
> ---
>
> Key: BEAM-5467
> URL: https://issues.apache.org/jira/browse/BEAM-5467
> Project: Beam
> Issue Type: Improvement
> Components: runner-flink
>Reporter: Thomas Weise
>Priority: Critical
> Labels: flake, portability-flink
> Time Spent: 17h 10m
> Remaining Estimate: 0h
>
>
> The job is currently flaky and does not provide a useful signal.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-5467) Python Flink ValidatesRunner job issues
[ https://issues.apache.org/jira/browse/BEAM-5467?focusedWorklogId=179530&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179530 ] ASF GitHub Bot logged work on BEAM-5467: Author: ASF GitHub Bot Created on: 28/Dec/18 18:31 Start Date: 28/Dec/18 18:31 Worklog Time Spent: 10m Work Description: mxm commented on issue #7376: [BEAM-5467] Increase test timeout for portable ValidatesRunner tests URL: https://github.com/apache/beam/pull/7376#issuecomment-450405936 Run Python Flink ValidatesRunner This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179530) Time Spent: 17.5h (was: 17h 20m) > Python Flink ValidatesRunner job issues > --- > > Key: BEAM-5467 > URL: https://issues.apache.org/jira/browse/BEAM-5467 > Project: Beam > Issue Type: Improvement > Components: runner-flink >Reporter: Thomas Weise >Priority: Critical > Labels: flake, portability-flink > Time Spent: 17.5h > Remaining Estimate: 0h > > > The job is currently flaky and does not provide a useful signal. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Commented] (BEAM-6289) Running a join on two Cassandra tables using FlinkRunner fails
[
https://issues.apache.org/jira/browse/BEAM-6289?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730412#comment-16730412
]
Shahar Frank commented on BEAM-6289:
Thanks for the info.
I will give these a try later on.
Just a question re. "Of course this wouldn't happen with a remote Flink
cluster." - The issue was first detected on a remote Flink cluster - I will
recheck it with this demo code - could this then be some other issue then?
And can you please tell me what the akka.ask.timeout value which is currently
being used? Is it very short? Because this entire code fail within a few
seconds - well below 10 seconds - so it's not that long - is it?
> Running a join on two Cassandra tables using FlinkRunner fails
> --
>
> Key: BEAM-6289
> URL: https://issues.apache.org/jira/browse/BEAM-6289
> Project: Beam
> Issue Type: Bug
> Components: io-java-cassandra, runner-flink
>Affects Versions: 2.8.0, 2.9.0
> Environment: Tested on Ubuntu 18
> Beam 2.8
> Tested with Flink:
> 1) [local]
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
> Tested using Cassandra [cqlsh 5.0.1 | Cassandra 3.11.3 | CQL spec 3.4.4 |
> Native protocol v4]:
> 1) In a local container
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
>Reporter: Shahar Frank
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: FlinkRunner, beam, bug, cassandra, flink, join
> Attachments: direct_runner_build.log, flink_runner_build.log
>
>
> Can't make a simple join on two Cassandra tables when using FlinkRunner.
> The same code works with a DirectRunner fails when used with FlinkRunner
> giving these (as well as many other) errors:
> {code:java}
> Caused by: akka.pattern.AskTimeoutException: Ask timed out on
> [Actor[akka://flink/user/dispatchere1f5abe7-6299-43ea-9182-24a2193e078f#-1757043920]]
> after [1 ms]. Sender[null] sent message of type
> "org.apache.flink.runtime.rpc.messages.LocalFencedMessage".
> at
> akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
> at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
> at
> scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
> at
> akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
> at java.lang.Thread.run(Thread.java:748)
> {code}
>
> The code can be found [here|https://github.com/srfrnk/beam-playground]
> Steps to reproduce:
> # Clone the repo to a linux (I;m on Ubuntu 18 but any *nix system would
> probably work - i.e. repl.it)
> # Follow the README to set up a Cassandra container + schema
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=flink > output/build.log
> 2>&1{code}
> to use FlinkRunner. See error in log at ./output/build.log
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=direct > output/build.log
> 2>&1{code}
> to use DirectRunner. See error in log at ./output/build.log
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-6282) Failure in post-release tests: Intra-project deps on vendored artifacts are incompatible with separate release
[
https://issues.apache.org/jira/browse/BEAM-6282?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730410#comment-16730410
]
Andrew Pilloud commented on BEAM-6282:
--
https://builds.apache.org/job/beam_PostRelease_NightlySnapshot/474/
> Failure in post-release tests: Intra-project deps on vendored artifacts are
> incompatible with separate release
> --
>
> Key: BEAM-6282
> URL: https://issues.apache.org/jira/browse/BEAM-6282
> Project: Beam
> Issue Type: Bug
> Components: build-system
>Reporter: Kenneth Knowles
>Assignee: Kenneth Knowles
>Priority: Critical
>
> https://builds.apache.org/job/beam_PostRelease_NightlySnapshot/467/console
> This is a test that runs wordcount against a snapshot release.
> {code}
> 06:03:45 [ERROR] Failed to execute goal
> org.codehaus.mojo:exec-maven-plugin:1.6.0:java (default-cli) on project
> word-count-beam: An exception occured while executing the Java class.
> org/apache/beam/vendor/grpc/v1p13p1/com/google/protobuf/MessageOrBuilder:
> org.apache.beam.vendor.grpc.v1p13p1.com.google.protobuf.MessageOrBuilder ->
> [Help 1]
> {code}
> {code}
> org/apache/beam/vendor/grpc/v1p13p1/com/google/protobuf/ProtocolMessageEnum:
> org.apache.beam.vendor.grpc.v1p13p1.com.google.protobuf.ProtocolMessageEnum
> {code}
> etc, etc. Still looking through the logs, but I am guessing the v1p13p1
> dependency is not available or not set up right.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-6316) beam-sdks-python:portableWordCountBatch flaky: could not load docker image
[
https://issues.apache.org/jira/browse/BEAM-6316?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730408#comment-16730408
]
Andrew Pilloud commented on BEAM-6316:
--
https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/108/
> beam-sdks-python:portableWordCountBatch flaky: could not load docker image
> --
>
> Key: BEAM-6316
> URL: https://issues.apache.org/jira/browse/BEAM-6316
> Project: Beam
> Issue Type: Bug
> Components: runner-flink, test-failures
>Reporter: Andrew Pilloud
>Assignee: Maximilian Michels
>Priority: Major
>
> https://builds.apache.org/job/beam_PreCommit_Portable_Python_Cron/105/
> {code}
> 10:08:30 INFO:root:Using latest locally built Python SDK docker image.
> 10:08:31 Unable to find image
> 'jenkins-docker-apache.bintray.io/beam/flink-job-server:latest' locally
> 10:08:31 INFO:root:Parsing unknown args: ['--environment_cache_millis=1']
> 10:08:31 docker: Error response from daemon: unknown: Repo 'apache' was not
> found.
> 10:08:31 See 'docker run --help'.
> 12:00:13 Build timed out (after 120 minutes). Marking the build as aborted.
> {code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-6280) Failure in PortableRunnerTest.test_error_traceback_includes_user_code
[
https://issues.apache.org/jira/browse/BEAM-6280?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730407#comment-16730407
]
Andrew Pilloud commented on BEAM-6280:
--
https://builds.apache.org/job/beam_PreCommit_Python_Cron/758/
> Failure in PortableRunnerTest.test_error_traceback_includes_user_code
> -
>
> Key: BEAM-6280
> URL: https://issues.apache.org/jira/browse/BEAM-6280
> Project: Beam
> Issue Type: Bug
> Components: sdk-py-core, test-failures
>Reporter: Kenneth Knowles
>Assignee: Robert Bradshaw
>Priority: Critical
>
> [https://builds.apache.org/job/beam_PreCommit_Python_Cron/732/]
> [https://builds.apache.org/job/beam_PreCommit_Python_Cron/732/testReport/apache_beam.runners.portability.portable_runner_test/PortableRunnerTest/test_error_traceback_includes_user_code/]
> [https://scans.gradle.com/s/do3hjulee3gaa/console-log?task=:beam-sdks-python:testPython3]
> {code:java}
> 'second' not found in 'Traceback (most recent call last):\n File
> "/home/jenkins/jenkins-slave/workspace/beam_PreCommit_Python_Cron/src/sdks/python/apache_beam/runners/portability/fn_api_runner_test.py",
> line 466, in test_error_traceback_includes_user_code\np |
> beam.Create([0]) | beam.Map(first) # pylint:
> disable=expression-not-assigned\n File
> "/home/jenkins/jenkins-slave/workspace/beam_PreCommit_Python_Cron/src/sdks/python/apache_beam/pipeline.py",
> line 425, in __exit__\nself.run().wait_until_finish()\n File
> "/home/jenkins/jenkins-slave/workspace/beam_PreCommit_Python_Cron/src/sdks/python/apache_beam/runners/portability/portable_runner.py",
> line 314, in wait_until_finish\nself._job_id, self._state,
> self._last_error_message()))\nRuntimeError: Pipeline
> job-cdcefe6d-1caa-4487-9e63-e971f67ec68c failed in state FAILED: start
> coder=WindowedValueCoder[BytesCoder], len(consumers)=1]]>\n'{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5662) [beam_PostCommit_Website_Publish] [:testWebsite] External link http://wiki.apache.org/incubator/BeamProposal failed: got a time out
[
https://issues.apache.org/jira/browse/BEAM-5662?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730405#comment-16730405
]
Andrew Pilloud commented on BEAM-5662:
--
https://builds.apache.org/job/beam_PreCommit_Website_Cron/472/
{code}
04:06:25 - ./generated-local-content/roadmap/go-sdk/index.html
04:06:25 * External link https://issues.apache.org/jira/browse/BEAM-2083
failed: response code 0 means something's wrong.
04:06:25 It's possible libcurl couldn't connect to the server or
perhaps the request timed out.
04:06:25 Sometimes, making too many requests at once also breaks
things.
04:06:25 Either way, the return message (if any) from the server
is: Server returned nothing (no headers, no data)
04:06:25 rake aborted!
{code}
> [beam_PostCommit_Website_Publish] [:testWebsite] External link
> http://wiki.apache.org/incubator/BeamProposal failed: got a time out
> ---
>
> Key: BEAM-5662
> URL: https://issues.apache.org/jira/browse/BEAM-5662
> Project: Beam
> Issue Type: Bug
> Components: test-failures
>Reporter: Scott Wegner
>Assignee: Scott Wegner
>Priority: Major
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> _Use this form to file an issue for test failure:_
> * [Jenkins
> Job|https://builds.apache.org/job/beam_PostCommit_Website_Publish/94/]
> * [Gradle Build
> Scan|https://scans.gradle.com/s/h4mayefon7v7q/console-log?task=:beam-website:testWebsite#L12]
> * [Test source
> code|https://github.com/apache/beam/blob/845f8d0abcc5a8d7f93457c27aff0feeb1a867d5/website/Rakefile#L6]
> Initial investigation:
> The failed link is http://wiki.apache.org/incubator/BeamProposal
> When I visit this link, it works for me. This is likely a flake.
>
> _After you've filled out the above details, please [assign the issue to an
> individual|https://beam.apache.org/contribute/postcommits-guides/index.html#find_specialist].
> Assignee should [treat test failures as
> high-priority|https://beam.apache.org/contribute/postcommits-policies/#assigned-failing-test],
> helping to fix the issue or find a more appropriate owner. See [Apache Beam
> Post-Commit
> Policies|https://beam.apache.org/contribute/postcommits-policies]._
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Commented] (BEAM-5215) GrpcDataServiceTest.testMessageReceivedBySingleClientWhenThereAreMultipleClients is flaky
[
https://issues.apache.org/jira/browse/BEAM-5215?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730404#comment-16730404
]
Andrew Pilloud commented on BEAM-5215:
--
https://builds.apache.org/job/beam_PreCommit_Java_Cron/756/
> GrpcDataServiceTest.testMessageReceivedBySingleClientWhenThereAreMultipleClients
> is flaky
> -
>
> Key: BEAM-5215
> URL: https://issues.apache.org/jira/browse/BEAM-5215
> Project: Beam
> Issue Type: Bug
> Components: runner-core
>Reporter: Andrew Pilloud
>Assignee: Luke Cwik
>Priority: Major
>
> [https://builds.apache.org/job/beam_PreCommit_Java_Cron/266/testReport/junit/org.apache.beam.runners.fnexecution.data/GrpcDataServiceTest/testMessageReceivedBySingleClientWhenThereAreMultipleClients/]
> {code:java}
> Stacktrace
> java.lang.AssertionError:
> Expected: iterable over [ instruction_reference: "0"
> target {
> primitive_transform_reference: "888"
> name: "test"
> }
> data: "\002A0\002B0\002C0"
> }
> data {
> instruction_reference: "0"
> target {
> primitive_transform_reference: "888"
> name: "test"
> }
> }
> >,instruction_reference: "1"
> target {
> primitive_transform_reference: "888"
> name: "test"
> }
> data: "\002A1\002B1\002C1"
> }
> data {
> instruction_reference: "1"
> target {
> primitive_transform_reference: "888"
> name: "test"
> }
> }
> >,instruction_reference: "2"
> target {
> primitive_transform_reference: "888"
> name: "test"
> }
> data: "\002A2\002B2\002C2"
> }
> data {
> instruction_reference: "2"
> target {
> primitive_transform_reference: "888"
> name: "test"
> }
> }
> >] in any order
> but:
> at org.hamcrest.MatcherAssert.assertThat(MatcherAssert.java:20)
> at org.junit.Assert.assertThat(Assert.java:956)
> at org.junit.Assert.assertThat(Assert.java:923)
> at
> org.apache.beam.runners.fnexecution.data.GrpcDataServiceTest.testMessageReceivedBySingleClientWhenThereAreMultipleClients(GrpcDataServiceTest.java:106)
> 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.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.runTestClass(JUnitTestClassExecutor.java:106)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:58)
> at
> org.gradle.api.internal.tasks.testing.junit.JUnitTestClassExecutor.execute(JUnitTestClassExecutor.java:38)
> at
> org.gradle.api.internal.tasks.testing.junit.AbstractJUnitTestClassProcessor.processTestClass(AbstractJUnitTestClassProcessor.java:66)
> at
> org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.processTestClass(SuiteTestClassProcessor.java:51)
> 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.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:35)
> at
> org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
> at
> org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:32)
> a
[jira] [Commented] (BEAM-6289) Running a join on two Cassandra tables using FlinkRunner fails
[
https://issues.apache.org/jira/browse/BEAM-6289?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16730390#comment-16730390
]
Maximilian Michels commented on BEAM-6289:
--
Thanks for the logs and for the great instructions on how to reproduce this.
Here's what happens:
The Cassandra source takes too long to generate the splits which is done at the
JobManager when the Beam job is sent to the cluster. This lets the job
submission time out because it does not receive the acknowledgement from the
JobManager until the splits are done. The timeout then results in a shutdown of
the local executor. Of course this wouldn't happen with a remote Flink cluster.
Two possibilities:
1. Increase the Flink setting for the Akka timeout, i.e. {{akka.ask.timeout: 1
minute}}. You will have to set the environment variable {{FLINK_CONF_DIR}} with
the directory containing the "flink-conf.yaml". Unfortunately, this will only
work in the latest master and upcoming release, as the config wasn't previously
loaded for local execution.
2. Change the splitting logic of the {{SourceInputFormat}} to perform early
splitting before job submission like we do in the streaming mode. This could
make sense for the next release.
For your test, you could consider adding the snapshot repo:
https://repository.apache.org/content/repositories/snapshots/ and using Beam
2.10.0.
> Running a join on two Cassandra tables using FlinkRunner fails
> --
>
> Key: BEAM-6289
> URL: https://issues.apache.org/jira/browse/BEAM-6289
> Project: Beam
> Issue Type: Bug
> Components: io-java-cassandra, runner-flink
>Affects Versions: 2.8.0, 2.9.0
> Environment: Tested on Ubuntu 18
> Beam 2.8
> Tested with Flink:
> 1) [local]
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
> Tested using Cassandra [cqlsh 5.0.1 | Cassandra 3.11.3 | CQL spec 3.4.4 |
> Native protocol v4]:
> 1) In a local container
> 2) Cluster inside a K8S cluster on minikube
> 3) Cluster inside a K8S cluster on GCP
>Reporter: Shahar Frank
>Assignee: Maximilian Michels
>Priority: Critical
> Labels: FlinkRunner, beam, bug, cassandra, flink, join
> Attachments: direct_runner_build.log, flink_runner_build.log
>
>
> Can't make a simple join on two Cassandra tables when using FlinkRunner.
> The same code works with a DirectRunner fails when used with FlinkRunner
> giving these (as well as many other) errors:
> {code:java}
> Caused by: akka.pattern.AskTimeoutException: Ask timed out on
> [Actor[akka://flink/user/dispatchere1f5abe7-6299-43ea-9182-24a2193e078f#-1757043920]]
> after [1 ms]. Sender[null] sent message of type
> "org.apache.flink.runtime.rpc.messages.LocalFencedMessage".
> at
> akka.pattern.PromiseActorRef$$anonfun$1.apply$mcV$sp(AskSupport.scala:604)
> at akka.actor.Scheduler$$anon$4.run(Scheduler.scala:126)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.unbatchedExecute(Future.scala:601)
> at
> scala.concurrent.BatchingExecutor$class.execute(BatchingExecutor.scala:109)
> at
> scala.concurrent.Future$InternalCallbackExecutor$.execute(Future.scala:599)
> at
> akka.actor.LightArrayRevolverScheduler$TaskHolder.executeTask(LightArrayRevolverScheduler.scala:329)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.executeBucket$1(LightArrayRevolverScheduler.scala:280)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.nextTick(LightArrayRevolverScheduler.scala:284)
> at
> akka.actor.LightArrayRevolverScheduler$$anon$4.run(LightArrayRevolverScheduler.scala:236)
> at java.lang.Thread.run(Thread.java:748)
> {code}
>
> The code can be found [here|https://github.com/srfrnk/beam-playground]
> Steps to reproduce:
> # Clone the repo to a linux (I;m on Ubuntu 18 but any *nix system would
> probably work - i.e. repl.it)
> # Follow the README to set up a Cassandra container + schema
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=flink > output/build.log
> 2>&1{code}
> to use FlinkRunner. See error in log at ./output/build.log
> # Run with
> {code}
> gradle --console=plain join-from-cassandra -Drunner=direct > output/build.log
> 2>&1{code}
> to use DirectRunner. See error in log at ./output/build.log
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6319) BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
[
https://issues.apache.org/jira/browse/BEAM-6319?focusedWorklogId=179524&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179524
]
ASF GitHub Bot logged work on BEAM-6319:
Author: ASF GitHub Bot
Created on: 28/Dec/18 17:18
Start Date: 28/Dec/18 17:18
Worklog Time Spent: 10m
Work Description: apilloud commented on issue #7368:
[BEAM-6319][BEAM-6312][BEAM-6311] Disable BigQueryIo validation in
BigQueryToTableIT
URL: https://github.com/apache/beam/pull/7368#issuecomment-450394834
According to the BigQuery docs, [table creation is eventually
consistent.](https://cloud.google.com/bigquery/troubleshooting-errors#metadata-errors-for-streaming-inserts)
It seems like this fix is just moving the race from pipeline creation to
worker setup. Does that code do something different that would allow it to
retry on this failure?
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179524)
Time Spent: 10m
Remaining Estimate: 0h
> BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom flaky
> ---
>
> Key: BEAM-6319
> URL: https://issues.apache.org/jira/browse/BEAM-6319
> Project: Beam
> Issue Type: Bug
> Components: io-java-gcp
>Reporter: Andrew Pilloud
>Assignee: Boyuan Zhang
>Priority: Major
> Time Spent: 10m
> Remaining Estimate: 0h
>
> https://builds.apache.org/job/beam_PostCommit_Java/2213/testReport/junit/org.apache.beam.sdk.io.gcp.bigquery/BigQueryToTableIT/testNewTypesQueryWithoutReshuffleWithCustom/
> {code}
> java.lang.IllegalArgumentException: Validation of query "SELECT bytes, date,
> time FROM [apache-beam-testing:bq_query_to_table_1545892541359_421.types]"
> failed. If the query depends on an earlier stage of the pipeline, This
> validation can be disabled using #withoutValidation.
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$TypedRead.validate(BigQueryIO.java:705)
> at
> org.apache.beam.sdk.Pipeline$ValidateVisitor.enterCompositeTransform(Pipeline.java:641)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:645)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
> at
> org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:311)
> at
> org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:245)
> at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
> at org.apache.beam.sdk.Pipeline.validate(Pipeline.java:577)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:312)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.runBigQueryToTablePipeline(BigQueryToTableIT.java:111)
> at
> org.apache.beam.sdk.io.gcp.bigquery.BigQueryToTableIT.testNewTypesQueryWithoutReshuffleWithCustom(BigQueryToTableIT.java:307)
> 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.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.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:349)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> at org.junit.runners.ParentRunner$3.run(ParentRunner.java:314)
> at org.junit.runners.P
[jira] [Work logged] (BEAM-6325) Error message is printed as bytes when failing to cross-compile
[
https://issues.apache.org/jira/browse/BEAM-6325?focusedWorklogId=179523&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179523
]
ASF GitHub Bot logged work on BEAM-6325:
Author: ASF GitHub Bot
Created on: 28/Dec/18 16:56
Start Date: 28/Dec/18 16:56
Worklog Time Spent: 10m
Work Description: lostluck commented on issue #7375: [BEAM-6325] Cast
cross compile output from []byte to string for printing
URL: https://github.com/apache/beam/pull/7375#issuecomment-450391261
R: @apilloud
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 179523)
Time Spent: 20m (was: 10m)
> Error message is printed as bytes when failing to cross-compile
> ---
>
> Key: BEAM-6325
> URL: https://issues.apache.org/jira/browse/BEAM-6325
> Project: Beam
> Issue Type: Bug
> Components: sdk-go
>Reporter: Ivar Gaitan
>Assignee: Robert Burke
>Priority: Minor
> Time Spent: 20m
> Remaining Estimate: 0h
>
> In {{beam/runners/universal/runnerlib/compile.go}}:
> If cross-compile fails in {{BuildWorkerBinary()}} the error message is
> printed as bytes. This is because {{Cmd.CombinedOutput()}} returns a byte
> slice as its output, which is then formatted as it's default value.
> {{if out, err := cmd.CombinedOutput(); err != nil {}}
> {{ return fmt.Errorf("failed to cross-compile %v: %v\n%v", program, err,
> out)}}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Work logged] (BEAM-6325) Error message is printed as bytes when failing to cross-compile
[ https://issues.apache.org/jira/browse/BEAM-6325?focusedWorklogId=179522&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179522 ] ASF GitHub Bot logged work on BEAM-6325: Author: ASF GitHub Bot Created on: 28/Dec/18 16:55 Start Date: 28/Dec/18 16:55 Worklog Time Spent: 10m Work Description: lostluck commented on pull request #7375: [BEAM-6325] Cast cross compile output from []byte to string for printing URL: https://github.com/apache/beam/pull/7375 Fix an oversight in the cross compile mechanism for the universal runner. Follow this checklist to help us incorporate your contribution quickly and easily: - [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf). It will help us expedite review of your Pull Request if you tag someone (e.g. `@username`) to look at it. Post-Commit Tests Status (on master branch) Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark --- | --- | --- | --- | --- | --- | --- | --- Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | --- | --- | --- | --- Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/) [](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/) Python | [](https://builds.apache.org/job/beam_PostCommit_Python_Verify/lastCompletedBuild/) | --- | [](https://builds.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/) [](https://builds.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | --- | --- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179522) Time Spent: 10m Remaining Estimate: 0h > Error message is printed as bytes when failing to cross-compile > --- > > Key: BEAM-6325 > URL: https://issues.apache.org/jira/brows
[jira] [Work logged] (BEAM-5985) Create jenkins jobs to run the load tests for Java SDK
[ https://issues.apache.org/jira/browse/BEAM-5985?focusedWorklogId=179520&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179520 ] ASF GitHub Bot logged work on BEAM-5985: Author: ASF GitHub Bot Created on: 28/Dec/18 16:41 Start Date: 28/Dec/18 16:41 Worklog Time Spent: 10m Work Description: kkucharc commented on issue #7184: [BEAM-5985] Create jenkins jobs to run the load tests for Java SDK URL: https://github.com/apache/beam/pull/7184#issuecomment-450388773 Run seed job This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179520) Time Spent: 9h (was: 8h 50m) > Create jenkins jobs to run the load tests for Java SDK > -- > > Key: BEAM-5985 > URL: https://issues.apache.org/jira/browse/BEAM-5985 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Lukasz Gajowy >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 9h > Remaining Estimate: 0h > > How/how often/in what cases we run those tests is yet to be decided (this is > part of the task) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-6155) Migrate the Go SDK to the modern GCS library
[ https://issues.apache.org/jira/browse/BEAM-6155?focusedWorklogId=179521&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-179521 ] ASF GitHub Bot logged work on BEAM-6155: Author: ASF GitHub Bot Created on: 28/Dec/18 16:46 Start Date: 28/Dec/18 16:46 Worklog Time Spent: 10m Work Description: lostluck commented on issue #7292: [BEAM-6155] Plumb the contexts though the gcsx library. URL: https://github.com/apache/beam/pull/7292#issuecomment-450389637 LGTM @aaltay Please merge at your convenience. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 179521) Time Spent: 1h (was: 50m) > Migrate the Go SDK to the modern GCS library > > > Key: BEAM-6155 > URL: https://issues.apache.org/jira/browse/BEAM-6155 > Project: Beam > Issue Type: Improvement > Components: sdk-go >Reporter: Andrew Brampton >Priority: Major > Time Spent: 1h > Remaining Estimate: 0h > > The gcsx package is using the google.golang.org/api/storage/v1 GCS library. > That library has been deprecated for ~6 months, and the recommendation is to > use the newer > [cloud.google.com/go/storage|https://godoc.org/cloud.google.com/go/storage] > package. That package supports newer features, and has built in connection > pooling, timeout support, retry with exponential backoff, etc. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work started] (BEAM-5985) Create jenkins jobs to run the load tests for Java SDK
[ https://issues.apache.org/jira/browse/BEAM-5985?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Work on BEAM-5985 started by Kasia Kucharczyk. -- > Create jenkins jobs to run the load tests for Java SDK > -- > > Key: BEAM-5985 > URL: https://issues.apache.org/jira/browse/BEAM-5985 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Lukasz Gajowy >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 8h 50m > Remaining Estimate: 0h > > How/how often/in what cases we run those tests is yet to be decided (this is > part of the task) -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5993) Create SideInput Load test
[ https://issues.apache.org/jira/browse/BEAM-5993?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kasia Kucharczyk resolved BEAM-5993. Resolution: Fixed Fix Version/s: 2.10.0 > Create SideInput Load test > -- > > Key: BEAM-5993 > URL: https://issues.apache.org/jira/browse/BEAM-5993 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Fix For: 2.10.0 > > Time Spent: 1h 40m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-6115) SyntheticSource bundle size parameter sometimes is casted to invalid type
[
https://issues.apache.org/jira/browse/BEAM-6115?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kasia Kucharczyk resolved BEAM-6115.
Resolution: Fixed
Fix Version/s: 2.9.0
> SyntheticSource bundle size parameter sometimes is casted to invalid type
> -
>
> Key: BEAM-6115
> URL: https://issues.apache.org/jira/browse/BEAM-6115
> Project: Beam
> Issue Type: Bug
> Components: testing
>Reporter: Kasia Kucharczyk
>Assignee: Kasia Kucharczyk
>Priority: Minor
> Fix For: 2.9.0
>
> Time Spent: 40m
> Remaining Estimate: 0h
>
> The parameter {code}bundle_size_in_elements{code} in SyntheticSources in
> Python in specific situations becomes `float` instead of `int` what causes
> failure on Dataflow:
> {code:java}
> Traceback (most recent call last):
> File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/batchworker.py",
> line 642, in do_work
> work_executor.execute()
> File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py",
> line 198, in execute
> self._split_task)
> File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py",
> line 206, in _perform_source_split_considering_api_limits
> desired_bundle_size)
> File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py",
> line 243, in _perform_source_split
> for split in source.split(desired_bundle_size):
> File
> "/usr/local/lib/python2.7/dist-packages/apache_beam/testing/synthetic_pipeline.py",
> line 222, in split
> bundle_size_in_elements):
> TypeError: range() integer step argument expected, got float.{code}
>
> Debugging showed that on Dataflow following line causes this problem (line
> 213-214):
> {code:python}max(1, self._num_records /
> self._initial_splitting_num_bundles){code}.
> In line 218, there is:
> {code:python}math.floor(math.sqrt(self._num_records)){code} which also
> returns float.
> In 222 line _bundle_size_in_elements_ is used to _range_ method which
> requires _int_.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Closed] (BEAM-5994) Publish metrics from load tests to BigQuery database
[ https://issues.apache.org/jira/browse/BEAM-5994?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kasia Kucharczyk closed BEAM-5994. -- Resolution: Fixed Fix Version/s: Not applicable > Publish metrics from load tests to BigQuery database > > > Key: BEAM-5994 > URL: https://issues.apache.org/jira/browse/BEAM-5994 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Fix For: Not applicable > > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-5778) Add integrations of Metrics API to Big Query for SyntheticSources load tests in Python SDK
[ https://issues.apache.org/jira/browse/BEAM-5778?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kasia Kucharczyk resolved BEAM-5778. Resolution: Fixed Fix Version/s: 2.9.0 > Add integrations of Metrics API to Big Query for SyntheticSources load tests > in Python SDK > -- > > Key: BEAM-5778 > URL: https://issues.apache.org/jira/browse/BEAM-5778 > Project: Beam > Issue Type: Improvement > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Fix For: 2.9.0 > > Time Spent: 4h 10m > Remaining Estimate: 0h > > Right now Metrics API collects basic metrics of load tests of > SyntheticSources (Python SDK). It should be collected in BigQuery for > presenting it on performance dashboards. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-4836) IOIT tests fails on Jenkins because of numpy version
[ https://issues.apache.org/jira/browse/BEAM-4836?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kasia Kucharczyk resolved BEAM-4836. Resolution: Fixed Fix Version/s: 2.6.0 > IOIT tests fails on Jenkins because of numpy version > > > Key: BEAM-4836 > URL: https://issues.apache.org/jira/browse/BEAM-4836 > Project: Beam > Issue Type: Bug > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Fix For: 2.6.0 > > Time Spent: 2h > Remaining Estimate: 0h > > Perfkit requires 1.13 numpy version. But probably because of > [https://github.com/apache/beam/pull/5565] in beam there is numpy-1.14. > Failure details (more details here > [https://builds.apache.org/job/beam_PerformanceTests_MongoDBIO_IT/452/] ): > perfkitbenchmarker.errors.PythonPackageRequirementUnfulfilled: A Python > package requirement was not met while checking > "/home/jenkins/jenkins-slave/workspace/beam_PerformanceTests_MongoDBIO_IT/PerfKitBenchmarker/requirements.txt": > numpy 1.14.5 is installed but numpy==1.13.3 is required -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Resolved] (BEAM-4414) Create more specific namespace for each IOIT in FileBasedIOIT
[ https://issues.apache.org/jira/browse/BEAM-4414?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kasia Kucharczyk resolved BEAM-4414. Resolution: Fixed Fix Version/s: 2.6.0 > Create more specific namespace for each IOIT in FileBasedIOIT > - > > Key: BEAM-4414 > URL: https://issues.apache.org/jira/browse/BEAM-4414 > Project: Beam > Issue Type: Bug > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Fix For: 2.6.0 > > Time Spent: 8h 10m > Remaining Estimate: 0h > > After changing namespaces (https://issues.apache.org/jira/browse/BEAM-4371) > file-based tests started failing because shared of namespace. All those tests > (e.g TextIOIT or AvroIOIT) should have specified 'subname' passed to > namespace. E.g. 'filebasedioithdfs-203' > 'filebasedioithdfs-text-203' -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Updated] (BEAM-4371) Namespace shares the same name between Jenkins Performance test jobs
[
https://issues.apache.org/jira/browse/BEAM-4371?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kasia Kucharczyk updated BEAM-4371:
---
Fix Version/s: (was: 2.4.0)
2.5.0
> Namespace shares the same name between Jenkins Performance test jobs
>
>
> Key: BEAM-4371
> URL: https://issues.apache.org/jira/browse/BEAM-4371
> Project: Beam
> Issue Type: Bug
> Components: testing
>Reporter: Kasia Kucharczyk
>Assignee: Kasia Kucharczyk
>Priority: Major
> Fix For: 2.5.0
>
> Time Spent: 10h 20m
> Remaining Estimate: 0h
>
> In _common_job_properties.groovy_ is defined _getKubernetesNamespace(def
> testName)_ which runs only after _seed job_.
> Expected behaviour is to create a new namespace with every test run.
> The failure was observed after retrying JdbcIOIT job. The first failure did
> not deleted namespace. Next jobs failed because of:
> {code}
> + kubectl
> --kubeconfig=/home/jenkins/jenkins-slave/workspace/beam_PerformanceTests_JDBC/config-jdbcioit-1526904136153
> create namespace jdbcioit-1526904136153 Error from server (AlreadyExists):
> namespaces "jdbcioit-1526904136153" already exists
> {code}
> Here are examples of jobs on Jenkins:
> [https://builds.apache.org/view/A-D/view/Beam/job/beam_PerformanceTests_JDBC/601/console]
> and
> [https://builds.apache.org/view/A-D/view/Beam/job/beam_PerformanceTests_JDBC/602/console]
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Resolved] (BEAM-4371) Namespace shares the same name between Jenkins Performance test jobs
[
https://issues.apache.org/jira/browse/BEAM-4371?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Kasia Kucharczyk resolved BEAM-4371.
Resolution: Fixed
Fix Version/s: 2.4.0
> Namespace shares the same name between Jenkins Performance test jobs
>
>
> Key: BEAM-4371
> URL: https://issues.apache.org/jira/browse/BEAM-4371
> Project: Beam
> Issue Type: Bug
> Components: testing
>Reporter: Kasia Kucharczyk
>Assignee: Kasia Kucharczyk
>Priority: Major
> Fix For: 2.4.0
>
> Time Spent: 10h 20m
> Remaining Estimate: 0h
>
> In _common_job_properties.groovy_ is defined _getKubernetesNamespace(def
> testName)_ which runs only after _seed job_.
> Expected behaviour is to create a new namespace with every test run.
> The failure was observed after retrying JdbcIOIT job. The first failure did
> not deleted namespace. Next jobs failed because of:
> {code}
> + kubectl
> --kubeconfig=/home/jenkins/jenkins-slave/workspace/beam_PerformanceTests_JDBC/config-jdbcioit-1526904136153
> create namespace jdbcioit-1526904136153 Error from server (AlreadyExists):
> namespaces "jdbcioit-1526904136153" already exists
> {code}
> Here are examples of jobs on Jenkins:
> [https://builds.apache.org/view/A-D/view/Beam/job/beam_PerformanceTests_JDBC/601/console]
> and
> [https://builds.apache.org/view/A-D/view/Beam/job/beam_PerformanceTests_JDBC/602/console]
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
