[GitHub] [flink] rmetzger commented on a change in pull request #11430: [FLINK-16633][AZP] Fix builds without s3 credentials
rmetzger commented on a change in pull request #11430: [FLINK-16633][AZP] Fix
builds without s3 credentials
URL: https://github.com/apache/flink/pull/11430#discussion_r394138116
##
File path: tools/azure-pipelines/jobs-template.yml
##
@@ -123,9 +123,9 @@ jobs:
- script: STAGE=test ${{parameters.environment}} ./tools/azure_controller.sh
$(module)
displayName: Test - $(module)
env:
- IT_CASE_S3_BUCKET: $(IT_CASE_S3_BUCKET)
- IT_CASE_S3_ACCESS_KEY: $(IT_CASE_S3_ACCESS_KEY)
- IT_CASE_S3_SECRET_KEY: $(IT_CASE_S3_SECRET_KEY)
+ IT_CASE_S3_BUCKET: ${{variables['IT_CASE_S3_BUCKET']}}
Review comment:
Yes :( I did not test this w/o credentials set.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r388031349
##
File path: docs/_includes/generated/rest_v1_dispatcher.html
##
@@ -4128,3 +4128,116 @@
+
+
+
+/taskmanagers/:taskmanagerid/logs
+
+
+Verb: GET
+Response code: 200 OK
+
+
+Provides access to task manager logs list.
+
+
+Path parameters
+
+
+
+
+taskmanagerid - 32-character hexadecimal
string that identifies a task manager.
+
+
+
+
+
+Request
+
+
+
+{}
+
+
+
+
+
+
+Response
+
+
+
+{
+ "type" : "object",
+ "id" :
"urn:jsonschema:org:apache:flink:runtime:rest:messages:taskmanager:LogsInfo",
+ "properties" : {
+"logs" : {
+ "type" : "array",
+ "items" : {
+"type" : "object",
+"id" :
"urn:jsonschema:org:apache:flink:runtime:rest:messages:taskmanager:LogInfo",
+"properties" : {
+ "name" : {
+"type" : "string"
+ },
+ "size" : {
+"type" : "long"
+ }
+}
+ }
+}
+ }
+}
+
+
+
+
+
+
+
+
+
+/taskmanagers/:taskmanagerid/log/:filename
+
Review comment:
As `/taskmanagers/:taskmanagerid/log` don't define in here, should this URL
define in here.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Closed] (FLINK-9142) Lower the minimum number of buffers for incoming channels to 1
[
https://issues.apache.org/jira/browse/FLINK-9142?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski closed FLINK-9142.
-
Fix Version/s: (was: 1.7.3)
Resolution: Duplicate
I'm closing this ticket, as it was duplicated and superseded by FLINK-16012 and
this looks inactive.
> Lower the minimum number of buffers for incoming channels to 1
> --
>
> Key: FLINK-9142
> URL: https://issues.apache.org/jira/browse/FLINK-9142
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Network
>Affects Versions: 1.5.0, 1.6.0
>Reporter: Nico Kruber
>Assignee: boshu Zheng
>Priority: Major
> Labels: pull-request-available
> Time Spent: 20m
> Remaining Estimate: 0h
>
> Even if we make the floating buffers optional, we still require
> {{taskmanager.network.memory.buffers-per-channel}} number of (exclusive)
> buffers per incoming channel with credit-based flow control while without,
> the minimum was 1 and only the maximum number of buffers was influenced by
> this parameter.
> {{taskmanager.network.memory.buffers-per-channel}} is set to {{2}} by default
> with the argumentation that this way we will have one buffer available for
> netty to process while a worker thread is processing/deserializing the other
> buffer. While this seems reasonable, it does increase our minimum
> requirements. Instead, we could probably live with {{1}} exclusive buffer and
> up to {{gate.getNumberOfInputChannels() * (networkBuffersPerChannel - 1) +
> extraNetworkBuffersPerGate}} floating buffers. That way we will have the same
> memory footprint as before with only slightly changed behaviour.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] pnowojski closed pull request #7579: [FLINK-9142] Lower the minimum number of buffers for incoming channels to 1
pnowojski closed pull request #7579: [FLINK-9142] Lower the minimum number of buffers for incoming channels to 1 URL: https://github.com/apache/flink/pull/7579 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] pnowojski commented on issue #7579: [FLINK-9142] Lower the minimum number of buffers for incoming channels to 1
pnowojski commented on issue #7579: [FLINK-9142] Lower the minimum number of buffers for incoming channels to 1 URL: https://github.com/apache/flink/pull/7579#issuecomment-600454958 I'm closing this PR, as it seems abandoned and there is an ongoing similar effort tracked in https://issues.apache.org/jira/browse/FLINK-16012 Please @kisimple let us know if we are missing something or if you would like to be involved in the new efforts. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log
list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394133992
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogsHandler.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.runtime.rest.handler.taskmanager;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import
org.apache.flink.runtime.resourcemanager.exceptions.UnknownTaskExecutorException;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.taskmanager.LogInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.LogsInfo;
+import
org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerIdPathParameter;
+import
org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMessageParameters;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.stream.Collectors;
+
+/**
+ * Handler which serves detailed TaskManager log list information.
+ */
+public class TaskManagerLogsHandler extends
AbstractTaskManagerHandler {
+
+ private final GatewayRetriever
resourceManagerGatewayRetriever;
+
+ public TaskManagerLogsHandler(
+ GatewayRetriever
leaderRetriever,
+ Time timeout,
+ Map responseHeaders,
+ MessageHeaders messageHeaders,
+ GatewayRetriever
resourceManagerGatewayRetriever) {
+ super(leaderRetriever, timeout, responseHeaders,
messageHeaders, resourceManagerGatewayRetriever);
+
+ this.resourceManagerGatewayRetriever =
Preconditions.checkNotNull(resourceManagerGatewayRetriever);
+ }
+
+ @Override
+ protected CompletableFuture handleRequest(
+ @Nonnull HandlerRequest request,
+ @Nonnull ResourceManagerGateway gateway) throws
RestHandlerException {
+ final ResourceID taskManagerId =
request.getPathParameter(TaskManagerIdPathParameter.class);
+ final ResourceManagerGateway resourceManagerGateway =
getResourceManagerGateway(resourceManagerGatewayRetriever);
+ final CompletableFuture>>
logsWithLengthFuture =
resourceManagerGateway.requestTaskManagerLogList(taskManagerId, timeout);
+
+ return logsWithLengthFuture.thenApply(logName2Sizes -> {
+ if (null != logName2Sizes) {
+ Collection logs =
logName2Sizes.stream().map(logName2Size -> new LogInfo(logName2Size.f0,
logName2Size.f1)).collect(Collectors.toSet());
+ return new LogsInfo(logs);
+ } else {
+ return LogsInfo.empty();
+ }
+ }).exceptionally(
+ (Throwable throwable) -> {
+ final Throwable strippedThrowable =
ExceptionUtils.stripExecutionException(throwable);
+
+ if (strippedThrowable instanceof
UnknownTaskExecutorException) {
+ throw new CompletionException(
+
[GitHub] [flink] GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log
list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394133992
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogsHandler.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.runtime.rest.handler.taskmanager;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import
org.apache.flink.runtime.resourcemanager.exceptions.UnknownTaskExecutorException;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.taskmanager.LogInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.LogsInfo;
+import
org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerIdPathParameter;
+import
org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMessageParameters;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.stream.Collectors;
+
+/**
+ * Handler which serves detailed TaskManager log list information.
+ */
+public class TaskManagerLogsHandler extends
AbstractTaskManagerHandler {
+
+ private final GatewayRetriever
resourceManagerGatewayRetriever;
+
+ public TaskManagerLogsHandler(
+ GatewayRetriever
leaderRetriever,
+ Time timeout,
+ Map responseHeaders,
+ MessageHeaders messageHeaders,
+ GatewayRetriever
resourceManagerGatewayRetriever) {
+ super(leaderRetriever, timeout, responseHeaders,
messageHeaders, resourceManagerGatewayRetriever);
+
+ this.resourceManagerGatewayRetriever =
Preconditions.checkNotNull(resourceManagerGatewayRetriever);
+ }
+
+ @Override
+ protected CompletableFuture handleRequest(
+ @Nonnull HandlerRequest request,
+ @Nonnull ResourceManagerGateway gateway) throws
RestHandlerException {
+ final ResourceID taskManagerId =
request.getPathParameter(TaskManagerIdPathParameter.class);
+ final ResourceManagerGateway resourceManagerGateway =
getResourceManagerGateway(resourceManagerGatewayRetriever);
+ final CompletableFuture>>
logsWithLengthFuture =
resourceManagerGateway.requestTaskManagerLogList(taskManagerId, timeout);
+
+ return logsWithLengthFuture.thenApply(logName2Sizes -> {
+ if (null != logName2Sizes) {
+ Collection logs =
logName2Sizes.stream().map(logName2Size -> new LogInfo(logName2Size.f0,
logName2Size.f1)).collect(Collectors.toSet());
+ return new LogsInfo(logs);
+ } else {
+ return LogsInfo.empty();
+ }
+ }).exceptionally(
+ (Throwable throwable) -> {
+ final Throwable strippedThrowable =
ExceptionUtils.stripExecutionException(throwable);
+
+ if (strippedThrowable instanceof
UnknownTaskExecutorException) {
+ throw new CompletionException(
+
[GitHub] [flink] GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log
list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394133406
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##
@@ -309,6 +311,25 @@ public TaskExecutor(
return
CompletableFuture.completedFuture(shuffleEnvironment.getPartitionsOccupyingLocalResources().isEmpty());
}
+ @Override
+ public CompletableFuture>>
requestLogList(Time timeout) {
+ final String logDir =
taskManagerConfiguration.getTaskManagerLogDir();
+ if (logDir != null) {
+ final File[] logFiles = new File(logDir).listFiles();
+
+ if (logFiles == null) {
+ return FutureUtils.completedExceptionally(
+ new FlinkException("The specific log
directory is not a valid directory."));
+ }
+
+ final List> logsWithLength =
Arrays.stream(logFiles)
+ .map(logFile -> Tuple2.of(logFile.getName(),
logFile.length()))
+ .collect(Collectors.toList());
+ return
CompletableFuture.completedFuture(logsWithLength);
+ }
+ return FutureUtils.completedExceptionally(new
FlinkException("There is no log file available on the TaskExecutor."));
Review comment:
Ok, that makes sense.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log
list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394133406
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##
@@ -309,6 +311,25 @@ public TaskExecutor(
return
CompletableFuture.completedFuture(shuffleEnvironment.getPartitionsOccupyingLocalResources().isEmpty());
}
+ @Override
+ public CompletableFuture>>
requestLogList(Time timeout) {
+ final String logDir =
taskManagerConfiguration.getTaskManagerLogDir();
+ if (logDir != null) {
+ final File[] logFiles = new File(logDir).listFiles();
+
+ if (logFiles == null) {
+ return FutureUtils.completedExceptionally(
+ new FlinkException("The specific log
directory is not a valid directory."));
+ }
+
+ final List> logsWithLength =
Arrays.stream(logFiles)
+ .map(logFile -> Tuple2.of(logFile.getName(),
logFile.length()))
+ .collect(Collectors.toList());
+ return
CompletableFuture.completedFuture(logsWithLength);
+ }
+ return FutureUtils.completedExceptionally(new
FlinkException("There is no log file available on the TaskExecutor."));
Review comment:
Ok, that's reasonable.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
GJL commented on a change in pull request #11250: [FLINK-16302][rest]add log
list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394133133
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##
@@ -1647,6 +1660,30 @@ private boolean isJobManagerConnectionValid(JobID
jobId, JobMasterId jobMasterId
return jmConnection != null &&
Objects.equals(jmConnection.getJobMasterId(), jobMasterId);
}
+ private CompletableFuture
requestFileUploadByFilePath(String filePath, Time timeout, String fileTag) {
Review comment:
I don't think we need it. This is a `private` method. The timeout argument
extraction happens before this method is invoked:
https://github.com/apache/flink/blob/ddec07323701a9998e854c378468617bafffa1f5/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaInvocationHandler.java#L306-L324
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * ec04379dc2b8e085cdd13aa552a97650dd31e65b Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153855376) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6343) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394129123
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogsHandler.java
##
@@ -0,0 +1,98 @@
+/*
+ * 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.flink.runtime.rest.handler.taskmanager;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import
org.apache.flink.runtime.resourcemanager.exceptions.UnknownTaskExecutorException;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.taskmanager.LogInfo;
+import org.apache.flink.runtime.rest.messages.taskmanager.LogsInfo;
+import
org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerIdPathParameter;
+import
org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMessageParameters;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.Preconditions;
+
+import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.stream.Collectors;
+
+/**
+ * Handler which serves detailed TaskManager log list information.
+ */
+public class TaskManagerLogsHandler extends
AbstractTaskManagerHandler {
+
+ private final GatewayRetriever
resourceManagerGatewayRetriever;
+
+ public TaskManagerLogsHandler(
+ GatewayRetriever
leaderRetriever,
+ Time timeout,
+ Map responseHeaders,
+ MessageHeaders messageHeaders,
+ GatewayRetriever
resourceManagerGatewayRetriever) {
+ super(leaderRetriever, timeout, responseHeaders,
messageHeaders, resourceManagerGatewayRetriever);
+
+ this.resourceManagerGatewayRetriever =
Preconditions.checkNotNull(resourceManagerGatewayRetriever);
+ }
+
+ @Override
+ protected CompletableFuture handleRequest(
+ @Nonnull HandlerRequest request,
+ @Nonnull ResourceManagerGateway gateway) throws
RestHandlerException {
+ final ResourceID taskManagerId =
request.getPathParameter(TaskManagerIdPathParameter.class);
+ final ResourceManagerGateway resourceManagerGateway =
getResourceManagerGateway(resourceManagerGatewayRetriever);
+ final CompletableFuture>>
logsWithLengthFuture =
resourceManagerGateway.requestTaskManagerLogList(taskManagerId, timeout);
+
+ return logsWithLengthFuture.thenApply(logName2Sizes -> {
+ if (null != logName2Sizes) {
+ Collection logs =
logName2Sizes.stream().map(logName2Size -> new LogInfo(logName2Size.f0,
logName2Size.f1)).collect(Collectors.toSet());
+ return new LogsInfo(logs);
+ } else {
+ return LogsInfo.empty();
+ }
+ }).exceptionally(
+ (Throwable throwable) -> {
+ final Throwable strippedThrowable =
ExceptionUtils.stripExecutionException(throwable);
+
+ if (strippedThrowable instanceof
UnknownTaskExecutorException) {
+ throw new CompletionException(
+
[GitHub] [flink] flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL
flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL URL: https://github.com/apache/flink/pull/11340#issuecomment-596050968 ## CI report: * 1e95f02cc2803e695eb28597de8f7344362826fd UNKNOWN * f38af5e48e36464e94a14cb7b10e7fc740081618 UNKNOWN * c7327ee006337e8ed7b4aab89ce4ba15aff0c576 UNKNOWN * e00910ef2a89260c19cd8cbaf2cb123dac0ed9f6 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153854408) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6342) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394124255
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
##
@@ -582,16 +582,41 @@ public void notifySlotAvailable(
}
@Override
- public CompletableFuture
requestTaskManagerFileUpload(ResourceID taskManagerId, FileType fileType, Time
timeout) {
- log.debug("Request file {} upload from TaskExecutor {}.",
fileType, taskManagerId);
+ public CompletableFuture
requestTaskManagerFileUploadByType(ResourceID taskManagerId, FileType fileType,
Time timeout) {
+ log.debug("Request file which type is {} upload from
TaskExecutor {}.", fileType, taskManagerId);
final WorkerRegistration taskExecutor =
taskExecutors.get(taskManagerId);
if (taskExecutor == null) {
- log.debug("Requested file {} upload from unregistered
TaskExecutor {}.", fileType, taskManagerId);
+ log.debug("Requested which type is {} upload from
unregistered TaskExecutor {}.", fileType, taskManagerId);
Review comment:
As here taskExecutor is null, so we need add unregistered for TaskExecutor.
```code
log.debug("Request upload of file {} from unregistered TaskExecutor {}.",
fileType, taskManagerId);
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394119984
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##
@@ -915,29 +936,21 @@ public void heartbeatFromResourceManager(ResourceID
resourceID) {
default:
filePath = null;
}
+ return requestFileUploadByFilePath(filePath, timeout,
fileType.toString());
+ }
- if (filePath != null && !filePath.isEmpty()) {
- final File file = new File(filePath);
-
- if (file.exists()) {
- final TransientBlobCache transientBlobService =
blobCacheService.getTransientBlobService();
- final TransientBlobKey transientBlobKey;
- try (FileInputStream fileInputStream = new
FileInputStream(file)) {
- transientBlobKey =
transientBlobService.putTransient(fileInputStream);
- } catch (IOException e) {
- log.debug("Could not upload file {}.",
fileType, e);
- return
FutureUtils.completedExceptionally(new FlinkException("Could not upload file "
+ fileType + '.', e));
- }
+ @Override
+ public CompletableFuture
requestFileUploadByName(String fileName, Time timeout) {
+ log.debug("Request file name is {} upload.", fileName);
Review comment:
We could add it in `requestFileUploadByFilePath().`
```code
log.debug("Received file upload request for file {}", fileTag);
```
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] becketqin commented on a change in pull request #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
becketqin commented on a change in pull request #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#discussion_r394119038 ## File path: flink-python/pyflink/ml/api/base.py ## @@ -265,11 +266,57 @@ def transform(self, t_env: TableEnvironment, input: Table) -> Table: return input def to_json(self) -> str: -import jsonpickle -return str(jsonpickle.encode(self, keys=True)) +""" +If all PipelineStages in this Pipeline are Java ones, this method will return a +Java json string, which can be loaded either from a Python Pipeline or a Java Pipeline, +otherwise, it returns a Python json string which can only be loaded from a Python Pipeline. +""" +# if all PipelineStages are Java ones, we use Java toJson() to generate Json string +# so that the string can also be loaded from Java side. +if all([type(stage) in [JavaTransformer, JavaEstimator, JavaModel] +for stage in self.get_stages()]): +j_pipeline = get_gateway().jvm.org.apache.flink.ml.api.core.Pipeline() +for stage in self.get_stages(): +stage._convert_params_to_java(stage._j_obj) +j_pipeline.appendStage(stage._j_obj) +return j_pipeline.toJson() +else: +import jsonpickle +return str(jsonpickle.encode(self, keys=True)) def load_json(self, json: str) -> None: -import jsonpickle -pipeline = jsonpickle.decode(json, keys=True) -for stage in pipeline.get_stages(): -self.append_stage(stage) +""" +This method can either load from a Java Pipeline json or a Python Pipeline json. +""" +# noinspection PyBroadException +try: +# try to load json with Python method +import jsonpickle +pipeline = jsonpickle.decode(json, keys=True) +for stage in pipeline.get_stages(): +self.append_stage(stage) +except Exception: Review comment: Do we know the exact exception type here? It would good to not naively catch all the exception if possible. If we don't know the exact exception, we should probably remember the exception and raise it when the Java Pipeline load also failed. The error message would then be something like: "Cannot load the JSON as either a Java Pipeline or a Python Pipeline. Python Pipeline load failed due to THIS_EXCEPTION. Java Pipeline load failed due to ANOTHER_EXCEPTION." This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL
flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL URL: https://github.com/apache/flink/pull/11340#issuecomment-596050968 ## CI report: * 1e95f02cc2803e695eb28597de8f7344362826fd UNKNOWN * f38af5e48e36464e94a14cb7b10e7fc740081618 UNKNOWN * c7327ee006337e8ed7b4aab89ce4ba15aff0c576 UNKNOWN * e00910ef2a89260c19cd8cbaf2cb123dac0ed9f6 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153854408) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6342) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * ec04379dc2b8e085cdd13aa552a97650dd31e65b Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153855376) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6343) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-599933380 ## CI report: * a50561c09abfe43b06ea42f987687d558fc18090 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153852420) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6341) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394117551
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##
@@ -1647,6 +1660,30 @@ private boolean isJobManagerConnectionValid(JobID
jobId, JobMasterId jobMasterId
return jmConnection != null &&
Objects.equals(jmConnection.getJobMasterId(), jobMasterId);
}
+ private CompletableFuture
requestFileUploadByFilePath(String filePath, Time timeout, String fileTag) {
Review comment:
This parameter is from the method in TaskExecutorGateway, and it's
RpcTimeout(**ps: Annotation for {@link RpcGateway} methods to specify an
additional timeout parameter for the returned future to be completed. The rest
of the provided parameters is passed to the remote rpc server for the rpc.**),
so we need it.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW merged pull request #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors
zhijiangW merged pull request #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors URL: https://github.com/apache/flink/pull/11400 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-10896) Extend state schema evolution support for more types
[
https://issues.apache.org/jira/browse/FLINK-10896?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061400#comment-17061400
]
Tzu-Li (Gordon) Tai commented on FLINK-10896:
-
[~alpinegizmo] [~stevenz3wu]
I think evolution for Scala case classes has indeed came up quite a few times.
Whether or not this is possible still needs some deeper investigation though.
I'm not aware of anyone working on this right now.
> Extend state schema evolution support for more types
>
>
> Key: FLINK-10896
> URL: https://issues.apache.org/jira/browse/FLINK-10896
> Project: Flink
> Issue Type: New Feature
> Components: API / Type Serialization System
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Major
>
> Whether or not a state's schema can be evolved (e.g., removing / adding
> fields to a POJO-typed state, modifying an Avro-type state's schema, etc.)
> depends on whether or not the type's corresponding {{TypeSerializer}} and its
> {{TypeSerializerSnapshot}} properly supports it.
> As of Flink 1.7, we currently only have support for evolving Avro types (with
> FLINK-10605).
> This tracks the support for other composite types that would benefit from an
> evolvable schema, such as POJOs, tuples, Scala case classes etc.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] zhijiangW commented on a change in pull request #11408: [FLINK-15989][FLINK-16225] Improve direct and metaspace out-of-memory error handling
zhijiangW commented on a change in pull request #11408:
[FLINK-15989][FLINK-16225] Improve direct and metaspace out-of-memory error
handling
URL: https://github.com/apache/flink/pull/11408#discussion_r394112080
##
File path:
flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
##
@@ -94,7 +99,15 @@ public static MemorySegment
allocateUnpooledOffHeapMemory(int size) {
* @return A new memory segment, backed by unpooled off-heap memory.
*/
public static MemorySegment allocateUnpooledOffHeapMemory(int size,
Object owner) {
- ByteBuffer memory = ByteBuffer.allocateDirect(size);
+ ByteBuffer memory;
+ try {
+ memory = ByteBuffer.allocateDirect(size);
+ } catch (OutOfMemoryError e) {
+ if (ExceptionUtils.isDirectOutOfMemoryError(e)) {
+
LOG.error(ExceptionUtils.TM_DIRECT_OOM_ERROR_MESSAGE, e);
+ }
+ throw e;
+ }
Review comment:
For the second usage, it had a ticket
(https://issues.apache.org/jira/browse/FLINK-15981) for improving this direct
memory overhead, not sure when will be done, just FYI.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11408: [FLINK-15989][FLINK-16225] Improve direct and metaspace out-of-memory error handling
zhijiangW commented on a change in pull request #11408:
[FLINK-15989][FLINK-16225] Improve direct and metaspace out-of-memory error
handling
URL: https://github.com/apache/flink/pull/11408#discussion_r394111632
##
File path:
flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java
##
@@ -94,7 +99,15 @@ public static MemorySegment
allocateUnpooledOffHeapMemory(int size) {
* @return A new memory segment, backed by unpooled off-heap memory.
*/
public static MemorySegment allocateUnpooledOffHeapMemory(int size,
Object owner) {
- ByteBuffer memory = ByteBuffer.allocateDirect(size);
+ ByteBuffer memory;
+ try {
+ memory = ByteBuffer.allocateDirect(size);
+ } catch (OutOfMemoryError e) {
+ if (ExceptionUtils.isDirectOutOfMemoryError(e)) {
+
LOG.error(ExceptionUtils.TM_DIRECT_OOM_ERROR_MESSAGE, e);
+ }
+ throw e;
+ }
Review comment:
Some inputs from my side: there are two usages for
`allocateUnpooledOffHeapMemory` ATM.
- One is for initializing `NetworkBufferPool` in `TaskManagerServices`, and
it also explicitly handles the `OutOfMemoryError` message inside. Maybe we also
want to unify the message there.
- Another is for reading blocking partitions for batch jobs on upstream
side, which is triggered by netty stack. This error would not be thrown
outsides to be visible in task stack, and be only transported to the downstream
side via network to cause the respective task failure. It also makes sense to
handle this special error on upstream side, to make processor exit if possible.
Otherwise the same error might happen again after failover recovery.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors
flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors URL: https://github.com/apache/flink/pull/11400#issuecomment-598611898 ## CI report: * a86f1e2e6785237893201dac302bc50041610d73 UNKNOWN * f0a936bd3411aa7ccaa7373fd51c0ed71a363629 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153850109) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6340) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * 1df1ef113620472a0ef36d75899bad699d22438f Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153843153) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6338) * ec04379dc2b8e085cdd13aa552a97650dd31e65b Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153855376) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6343) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-599933380 ## CI report: * a50561c09abfe43b06ea42f987687d558fc18090 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153852420) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6341) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * 5bfd373640181fb6a0a8b0276ef7d03e53927f72 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153845545) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * 1df1ef113620472a0ef36d75899bad699d22438f Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153843153) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6338) * ec04379dc2b8e085cdd13aa552a97650dd31e65b UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL
flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL URL: https://github.com/apache/flink/pull/11340#issuecomment-596050968 ## CI report: * 1e95f02cc2803e695eb28597de8f7344362826fd UNKNOWN * f38af5e48e36464e94a14cb7b10e7fc740081618 UNKNOWN * c7327ee006337e8ed7b4aab89ce4ba15aff0c576 UNKNOWN * 0805a77fddbe29c0e9704895bcb021ff76a0762e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/153669034) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6323) * e00910ef2a89260c19cd8cbaf2cb123dac0ed9f6 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153854408) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6342) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL
flinkbot edited a comment on issue #11340: [Flink 14338] Upgrade Calcite version to 1.22 for Flink SQL URL: https://github.com/apache/flink/pull/11340#issuecomment-596050968 ## CI report: * 1e95f02cc2803e695eb28597de8f7344362826fd UNKNOWN * f38af5e48e36464e94a14cb7b10e7fc740081618 UNKNOWN * c7327ee006337e8ed7b4aab89ce4ba15aff0c576 UNKNOWN * 0805a77fddbe29c0e9704895bcb021ff76a0762e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/153669034) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6323) * e00910ef2a89260c19cd8cbaf2cb123dac0ed9f6 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * 1df1ef113620472a0ef36d75899bad699d22438f Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153843153) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6338) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-599933380 ## CI report: * f5f6f03a435cdc7ee3a35818ed2422fc88c784e0 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153620246) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6313) * a50561c09abfe43b06ea42f987687d558fc18090 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153852420) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors
flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors URL: https://github.com/apache/flink/pull/11400#issuecomment-598611898 ## CI report: * a86f1e2e6785237893201dac302bc50041610d73 UNKNOWN * f0a936bd3411aa7ccaa7373fd51c0ed71a363629 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153850109) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6340) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * 5bfd373640181fb6a0a8b0276ef7d03e53927f72 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153845545) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Created] (FLINK-16642) CSV TableSource / TableSink shouldn't be in flink-table-api-java-bridge package
Kurt Young created FLINK-16642: -- Summary: CSV TableSource / TableSink shouldn't be in flink-table-api-java-bridge package Key: FLINK-16642 URL: https://issues.apache.org/jira/browse/FLINK-16642 Project: Flink Issue Type: Improvement Components: Table SQL / API Affects Versions: 1.10.0 Reporter: Kurt Young If user wants to use pure table api program to read/write csv data, he should be able to do this with only the following modules: # flink-table-common (for filesystem connector) # flink-table-api-java (for TableEnvironemtn) # flink-csv (for csv format) But right now it seems the user also has to depend on `flink-table-api-scala` because csv table source is in that module. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Assigned] (FLINK-16641) Announce sender's backlog to solve the deadlock issue without exclusive buffers
[ https://issues.apache.org/jira/browse/FLINK-16641?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Zhijiang reassigned FLINK-16641: Assignee: Yingjie Cao > Announce sender's backlog to solve the deadlock issue without exclusive > buffers > --- > > Key: FLINK-16641 > URL: https://issues.apache.org/jira/browse/FLINK-16641 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Network >Reporter: Zhijiang >Assignee: Yingjie Cao >Priority: Major > Fix For: 1.11.0 > > > This is the second ingredient besides FLINK-16404 to solve the deadlock > problem without exclusive buffers. > The scenario is as follows: > * The data in subpartition with positive backlog can be sent without doubt > because the exclusive credits would be feedback finally. > * Without exclusive buffers, the receiver would not request floating buffers > for 0 backlog. But when the new backlog is added into such subpartition, it > has no way to notify the receiver side without positive credits ATM. > * So it would result in waiting for each other between receiver and sender > sides to cause deadlock. The sender waits for credit to notify backlog and > the receiver waits for backlog to request floating credits. > To solve the above problem, the sender needs a separate message to announce > backlog sometimes besides existing `BufferResponse`. Then the receiver can > get this info to request floating buffers to feedback. > The side effect brought is to increase network transport delay and throughput > regression. We can measure how much it effects in existing micro-benchmark. > It might probably bear this effect to get a benefit of fast checkpoint > without exclusive buffers. We can give the proper explanations in respective > configuration options to let users make the final decision in practice. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (FLINK-16641) Announce sender's backlog to solve the deadlock issue without exclusive buffers
Zhijiang created FLINK-16641: Summary: Announce sender's backlog to solve the deadlock issue without exclusive buffers Key: FLINK-16641 URL: https://issues.apache.org/jira/browse/FLINK-16641 Project: Flink Issue Type: Sub-task Components: Runtime / Network Reporter: Zhijiang Fix For: 1.11.0 This is the second ingredient besides FLINK-16404 to solve the deadlock problem without exclusive buffers. The scenario is as follows: * The data in subpartition with positive backlog can be sent without doubt because the exclusive credits would be feedback finally. * Without exclusive buffers, the receiver would not request floating buffers for 0 backlog. But when the new backlog is added into such subpartition, it has no way to notify the receiver side without positive credits ATM. * So it would result in waiting for each other between receiver and sender sides to cause deadlock. The sender waits for credit to notify backlog and the receiver waits for backlog to request floating credits. To solve the above problem, the sender needs a separate message to announce backlog sometimes besides existing `BufferResponse`. Then the receiver can get this info to request floating buffers to feedback. The side effect brought is to increase network transport delay and throughput regression. We can measure how much it effects in existing micro-benchmark. It might probably bear this effect to get a benefit of fast checkpoint without exclusive buffers. We can give the proper explanations in respective configuration options to let users make the final decision in practice. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
[
https://issues.apache.org/jira/browse/FLINK-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061371#comment-17061371
]
Benchao Li commented on FLINK-16639:
+1 for using {{format.ignore-parse-errors}}.
FYI: It's introduced by https://issues.apache.org/jira/browse/FLINK-15396.
> Flink SQL Kafka source connector, add the no json format filter params when
> format.type is json
> ---
>
> Key: FLINK-16639
> URL: https://issues.apache.org/jira/browse/FLINK-16639
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Kafka
>Reporter: LakeShen
>Priority: Minor
>
> In my thought, kafka source connector is the one of most frequently used
> connector in flink sql. Flink sql kafka source connector supports the
> json,csv or other data format. But there is a problem for json format in
> kafka source connector. For example, flink sql kafka source ddl l
> like this:
> CREATE TABLE team_olap_table (
> a varchar,
> b varchar,
> )
> with (
> 'connector.type' = 'kafka',
> 'connector.version' = '0.10',
> 'connector.topic' = topics',
> 'connector.properties.0.key' = 'group.id',
> 'connector.properties.0.value' = 'hello_world',
> 'connector.properties.1.key' = 'bootstrap.servers',
> 'connector.properties.1.value' = 'xxx',
> 'connector.property-version' = '1',
> 'connector.startup-mode' = 'latest-offset',
> 'format.type' = 'json',
> 'format.property-version' = '1',
> 'format.derive-schema' = 'true',
> 'update-mode' = 'append'
> );
> If the kafka topic messages are not json format ,just one or two records,the
> flink sql task will fail-over all the time .
> In order to solve this problem , if flink sql source connector use the
> json-format, I want to add the 'format.fail-on-not-json-record' param in
> flink-json module, if this param is true（default），when read the no-json
> records, the flink will fail, if this param is false, the flink sql task will
> filter no-json records,the flink task running normally.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors
flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors URL: https://github.com/apache/flink/pull/11400#issuecomment-598611898 ## CI report: * a86f1e2e6785237893201dac302bc50041610d73 UNKNOWN * ea3d7681318c2a9dcdcb6a27cae8c2f0e95ea316 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153468221) * f0a936bd3411aa7ccaa7373fd51c0ed71a363629 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153850109) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6340) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#issuecomment-599596430 ## CI report: * acf17c53b9dd8dad8d8e2286c2140f96dca488b3 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153843138) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6337) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
flinkbot edited a comment on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-599933380 ## CI report: * f5f6f03a435cdc7ee3a35818ed2422fc88c784e0 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153620246) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6313) * a50561c09abfe43b06ea42f987687d558fc18090 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Updated] (FLINK-16404) Avoid caching buffers for blocked input channels before barrier alignment
[ https://issues.apache.org/jira/browse/FLINK-16404?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Zhijiang updated FLINK-16404: - Summary: Avoid caching buffers for blocked input channels before barrier alignment (was: Solve the potential deadlock problem when reducing exclusive buffers to zero) > Avoid caching buffers for blocked input channels before barrier alignment > - > > Key: FLINK-16404 > URL: https://issues.apache.org/jira/browse/FLINK-16404 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Network >Reporter: Zhijiang >Assignee: Yingjie Cao >Priority: Major > Labels: pull-request-available > Fix For: 1.11.0 > > Time Spent: 10m > Remaining Estimate: 0h > > One motivation of this issue is for reducing the in-flight data in the case > of back pressure to speed up checkpoint. The current default exclusive > buffers per channel is 2. If we reduce it to 0 and increase somewhat floating > buffers for compensation, it might cause deadlock problem because all the > floating buffers might be requested away by some blocked input channels and > never recycled until barrier alignment. > In order to solve above deadlock concern, we can make some logic changes on > both sender and receiver sides. > * Sender side: It should revoke previous received credit after sending > checkpoint barrier, that means it would not send any following buffers until > receiving new credits. > * Receiver side: The respective channel releases the requested floating > buffers if barrier is received from the network. After barrier alignment, it > would request floating buffers for the channels with positive backlog, and > notify the sender side of available credits. Then the sender can continue > transporting the buffers. > Based on above changes, we can also remove the `BufferStorage` component > completely, because the receiver would never reading buffers for blocked > channels. Another possible benefit is that the floating buffers might be more > properly made use of before barrier alignment. > The only side effect would bring somehow cold setup after barrier alignment. > That means the sender side has to wait for credit feedback to transport data > just after alignment, which would impact on delay and network throughput. But > considering the checkpoint interval not too short in general, so the above > side effect can be ignored in practice. We can further verify it via existing > micro-benchmark. > After this ticket done, we still can not set exclusive buffers to zero ATM, > there exists another deadlock issue which would be solved separately in > another ticket. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Updated] (FLINK-16404) Solve the potential deadlock problem when reducing exclusive buffers to zero
[ https://issues.apache.org/jira/browse/FLINK-16404?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Zhijiang updated FLINK-16404: - Description: One motivation of this issue is for reducing the in-flight data in the case of back pressure to speed up checkpoint. The current default exclusive buffers per channel is 2. If we reduce it to 0 and increase somewhat floating buffers for compensation, it might cause deadlock problem because all the floating buffers might be requested away by some blocked input channels and never recycled until barrier alignment. In order to solve above deadlock concern, we can make some logic changes on both sender and receiver sides. * Sender side: It should revoke previous received credit after sending checkpoint barrier, that means it would not send any following buffers until receiving new credits. * Receiver side: The respective channel releases the requested floating buffers if barrier is received from the network. After barrier alignment, it would request floating buffers for the channels with positive backlog, and notify the sender side of available credits. Then the sender can continue transporting the buffers. Based on above changes, we can also remove the `BufferStorage` component completely, because the receiver would never reading buffers for blocked channels. Another possible benefit is that the floating buffers might be more properly made use of before barrier alignment. The only side effect would bring somehow cold setup after barrier alignment. That means the sender side has to wait for credit feedback to transport data just after alignment, which would impact on delay and network throughput. But considering the checkpoint interval not too short in general, so the above side effect can be ignored in practice. We can further verify it via existing micro-benchmark. After this ticket done, we still can not set exclusive buffers to zero ATM, there exists another deadlock issue which would be solved separately in another ticket. was: One motivation of this issue is for reducing the in-flight data in the case of back pressure to speed up checkpoint. The current default exclusive buffers per channel is 2. If we reduce it to 0 and increase somewhat floating buffers for compensation, it might cause deadlock problem because all the floating buffers might be requested away by some blocked input channels and never recycled until barrier alignment. In order to solve above deadlock concern, we can make some logic changes on both sender and receiver sides. * Sender side: it should revoke previous received credit after sending checkpoint barrier, that means it would not send any following buffers until receiving new credits. * Receiver side: after processing the barrier from one channel and setting it blocked, it should release the available floating buffers for this blocked channel, and restore requesting floating buffers until barrier alignment. That means the receiver would only announce new credits to sender side after barrier alignment. Another possible benefit to do so is that the floating buffers might be more properly made use of before barrier alignment. We can further verify the performance concern via existing micro-benchmark. > Solve the potential deadlock problem when reducing exclusive buffers to zero > > > Key: FLINK-16404 > URL: https://issues.apache.org/jira/browse/FLINK-16404 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Network >Reporter: Zhijiang >Assignee: Yingjie Cao >Priority: Major > Labels: pull-request-available > Fix For: 1.11.0 > > Time Spent: 10m > Remaining Estimate: 0h > > One motivation of this issue is for reducing the in-flight data in the case > of back pressure to speed up checkpoint. The current default exclusive > buffers per channel is 2. If we reduce it to 0 and increase somewhat floating > buffers for compensation, it might cause deadlock problem because all the > floating buffers might be requested away by some blocked input channels and > never recycled until barrier alignment. > In order to solve above deadlock concern, we can make some logic changes on > both sender and receiver sides. > * Sender side: It should revoke previous received credit after sending > checkpoint barrier, that means it would not send any following buffers until > receiving new credits. > * Receiver side: The respective channel releases the requested floating > buffers if barrier is received from the network. After barrier alignment, it > would request floating buffers for the channels with positive backlog, and > notify the sender side of available credits. Then the sender can continue > transporting the buffers.
[jira] [Commented] (FLINK-16620) Add attempt information in logging
[
https://issues.apache.org/jira/browse/FLINK-16620?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061363#comment-17061363
]
Jiayi Liao commented on FLINK-16620:
[~roman_khachatryan] Totally agree with you. This comes to my mind when I was
trying to locate the problem in our production environment and some tasks
restarted several times, which made me have to identify the executions based on
the deploying timestamp in JM when reading logs.
> Add attempt information in logging
> --
>
> Key: FLINK-16620
> URL: https://issues.apache.org/jira/browse/FLINK-16620
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Task
>Affects Versions: 1.10.0
>Reporter: Jiayi Liao
>Priority: Minor
>
> Currently logging in places such as {{Task}} and {{StreamTask}} , is using
> {{taskNameWithSubtasks}} in {{TaskInfo}} to represent an execution. I think
> it'll be more user-friendly if we can add attempt information into the
> logging.
> The format can be consitent with logging information in {{Execution}} :
> {code:java}
> MySink (3/10) - execution #0
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] PatrickRen commented on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
PatrickRen commented on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-600405696 Any comment from Becket @becketqin ? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-16597) tryCleanupOnError should happen before close
[
https://issues.apache.org/jira/browse/FLINK-16597?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061361#comment-17061361
]
Jiayi Liao commented on FLINK-16597:
[~roman_khachatryan]
Behaviours in {{OutputFormatSinkFunction}} also makes sense to me. About your
two points:
1. {{DataSinkTask#invoke}}: "wrap exception thrown from close", do you mean
putting the 1st close call to the removed 2nd close call's location? Shouldn't
we call {{tryCleanupOnError}} when the exception is thrown from the new close
function?
2. {{DataSinkTask#cancel}}: I think {{cancel}} is also an error from user's
perspective because the work in format is not done yet. And I guess that why
Flink doesn't move {{tryCleanupOnError}} inside the catch block at the first
place.
> tryCleanupOnError should happen before close
>
>
> Key: FLINK-16597
> URL: https://issues.apache.org/jira/browse/FLINK-16597
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Task
>Affects Versions: 1.10.0
>Reporter: Jiayi Liao
>Priority: Major
>
> If {{DataSinkTask}} fails or throws an exception, the {{OutputFormat}} will
> call {{tryCleanupOnError}} before {{close}}. But the calling order is reverse
> when {{DataSinkTask}} is cancelled, which doesn't make much sense that we can
> still clean the output format after it's closed.
> I'm not very sure if this is a bug. But this does mislead our developers when
> implementing our customized {{OutputFormat}}. cc [~sewen]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors
flinkbot edited a comment on issue #11400: [FLINK-16586][network] Build ResultSubpartitionInfo and InputChannelInfo in respective constructors URL: https://github.com/apache/flink/pull/11400#issuecomment-598611898 ## CI report: * a86f1e2e6785237893201dac302bc50041610d73 UNKNOWN * ea3d7681318c2a9dcdcb6a27cae8c2f0e95ea316 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153468221) * f0a936bd3411aa7ccaa7373fd51c0ed71a363629 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335) * 5bfd373640181fb6a0a8b0276ef7d03e53927f72 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153845545) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6339) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * 1df1ef113620472a0ef36d75899bad699d22438f Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153843153) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6338) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#issuecomment-599596430 ## CI report: * acf17c53b9dd8dad8d8e2286c2140f96dca488b3 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153843138) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6337) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] PatrickRen commented on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
PatrickRen commented on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-600405311 Hi Jark @wuchong , I noticed that if we remove the line you mentioned, the variable `zk` will become unused: https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableTestBase.java#L69 But it is still referenced in part of commented code: https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableTestBase.java#L115 Should I remove the variable `zk`, or keep it in case of those commented code being used in the future? I'm not sure about the purpose of the commented code... This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Closed] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
[ https://issues.apache.org/jira/browse/FLINK-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kurt Young closed FLINK-16639. -- Fix Version/s: (was: 1.10.2) Resolution: Invalid > Flink SQL Kafka source connector, add the no json format filter params when > format.type is json > --- > > Key: FLINK-16639 > URL: https://issues.apache.org/jira/browse/FLINK-16639 > Project: Flink > Issue Type: Improvement > Components: Connectors / Kafka >Reporter: LakeShen >Priority: Minor > > In my thought, kafka source connector is the one of most frequently used > connector in flink sql. Flink sql kafka source connector supports the > json,csv or other data format. But there is a problem for json format in > kafka source connector. For example, flink sql kafka source ddl l > like this: > CREATE TABLE team_olap_table ( > a varchar, > b varchar, > ) > with ( > 'connector.type' = 'kafka', > 'connector.version' = '0.10', > 'connector.topic' = topics', > 'connector.properties.0.key' = 'group.id', > 'connector.properties.0.value' = 'hello_world', > 'connector.properties.1.key' = 'bootstrap.servers', > 'connector.properties.1.value' = 'xxx', > 'connector.property-version' = '1', > 'connector.startup-mode' = 'latest-offset', > 'format.type' = 'json', > 'format.property-version' = '1', > 'format.derive-schema' = 'true', > 'update-mode' = 'append' > ); > If the kafka topic messages are not json format ,just one or two records,the > flink sql task will fail-over all the time . > In order to solve this problem , if flink sql source connector use the > json-format, I want to add the 'format.fail-on-not-json-record' param in > flink-json module, if this param is true（default），when read the no-json > records, the flink will fail, if this param is false, the flink sql task will > filter no-json records,the flink task running normally. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Reopened] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
[ https://issues.apache.org/jira/browse/FLINK-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Kurt Young reopened FLINK-16639: > Flink SQL Kafka source connector, add the no json format filter params when > format.type is json > --- > > Key: FLINK-16639 > URL: https://issues.apache.org/jira/browse/FLINK-16639 > Project: Flink > Issue Type: Improvement > Components: Connectors / Kafka >Reporter: LakeShen >Priority: Minor > Fix For: 1.10.2 > > > In my thought, kafka source connector is the one of most frequently used > connector in flink sql. Flink sql kafka source connector supports the > json,csv or other data format. But there is a problem for json format in > kafka source connector. For example, flink sql kafka source ddl l > like this: > CREATE TABLE team_olap_table ( > a varchar, > b varchar, > ) > with ( > 'connector.type' = 'kafka', > 'connector.version' = '0.10', > 'connector.topic' = topics', > 'connector.properties.0.key' = 'group.id', > 'connector.properties.0.value' = 'hello_world', > 'connector.properties.1.key' = 'bootstrap.servers', > 'connector.properties.1.value' = 'xxx', > 'connector.property-version' = '1', > 'connector.startup-mode' = 'latest-offset', > 'format.type' = 'json', > 'format.property-version' = '1', > 'format.derive-schema' = 'true', > 'update-mode' = 'append' > ); > If the kafka topic messages are not json format ,just one or two records,the > flink sql task will fail-over all the time . > In order to solve this problem , if flink sql source connector use the > json-format, I want to add the 'format.fail-on-not-json-record' param in > flink-json module, if this param is true（default），when read the no-json > records, the flink will fail, if this param is false, the flink sql task will > filter no-json records,the flink task running normally. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335) * 5bfd373640181fb6a0a8b0276ef7d03e53927f72 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
[
https://issues.apache.org/jira/browse/FLINK-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061351#comment-17061351
]
Kurt Young commented on FLINK-16639:
does current \{{format.ignore-parse-errors}} meet your requirement?
> Flink SQL Kafka source connector, add the no json format filter params when
> format.type is json
> ---
>
> Key: FLINK-16639
> URL: https://issues.apache.org/jira/browse/FLINK-16639
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / Kafka
>Reporter: LakeShen
>Priority: Minor
> Fix For: 1.10.2
>
>
> In my thought, kafka source connector is the one of most frequently used
> connector in flink sql. Flink sql kafka source connector supports the
> json,csv or other data format. But there is a problem for json format in
> kafka source connector. For example, flink sql kafka source ddl l
> like this:
> CREATE TABLE team_olap_table (
> a varchar,
> b varchar,
> )
> with (
> 'connector.type' = 'kafka',
> 'connector.version' = '0.10',
> 'connector.topic' = topics',
> 'connector.properties.0.key' = 'group.id',
> 'connector.properties.0.value' = 'hello_world',
> 'connector.properties.1.key' = 'bootstrap.servers',
> 'connector.properties.1.value' = 'xxx',
> 'connector.property-version' = '1',
> 'connector.startup-mode' = 'latest-offset',
> 'format.type' = 'json',
> 'format.property-version' = '1',
> 'format.derive-schema' = 'true',
> 'update-mode' = 'append'
> );
> If the kafka topic messages are not json format ,just one or two records,the
> flink sql task will fail-over all the time .
> In order to solve this problem , if flink sql source connector use the
> json-format, I want to add the 'format.fail-on-not-json-record' param in
> flink-json module, if this param is true（default），when read the no-json
> records, the flink will fail, if this param is false, the flink sql task will
> filter no-json records,the flink task running normally.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Closed] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
[ https://issues.apache.org/jira/browse/FLINK-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] LakeShen closed FLINK-16639. Resolution: Fixed > Flink SQL Kafka source connector, add the no json format filter params when > format.type is json > --- > > Key: FLINK-16639 > URL: https://issues.apache.org/jira/browse/FLINK-16639 > Project: Flink > Issue Type: Improvement > Components: Connectors / Kafka >Reporter: LakeShen >Priority: Minor > Fix For: 1.10.2 > > > In my thought, kafka source connector is the one of most frequently used > connector in flink sql. Flink sql kafka source connector supports the > json,csv or other data format. But there is a problem for json format in > kafka source connector. For example, flink sql kafka source ddl l > like this: > CREATE TABLE team_olap_table ( > a varchar, > b varchar, > ) > with ( > 'connector.type' = 'kafka', > 'connector.version' = '0.10', > 'connector.topic' = topics', > 'connector.properties.0.key' = 'group.id', > 'connector.properties.0.value' = 'hello_world', > 'connector.properties.1.key' = 'bootstrap.servers', > 'connector.properties.1.value' = 'xxx', > 'connector.property-version' = '1', > 'connector.startup-mode' = 'latest-offset', > 'format.type' = 'json', > 'format.property-version' = '1', > 'format.derive-schema' = 'true', > 'update-mode' = 'append' > ); > If the kafka topic messages are not json format ,just one or two records,the > flink sql task will fail-over all the time . > In order to solve this problem , if flink sql source connector use the > json-format, I want to add the 'format.fail-on-not-json-record' param in > flink-json module, if this param is true（default），when read the no-json > records, the flink will fail, if this param is false, the flink sql task will > filter no-json records,the flink task running normally. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Assigned] (FLINK-13483) PrestoS3FileSystemITCase.testDirectoryListing fails on Travis
[
https://issues.apache.org/jira/browse/FLINK-13483?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yu Li reassigned FLINK-13483:
-
Assignee: Lu Niu
Assigned, thanks for volunteering [~qqibrow]!
> PrestoS3FileSystemITCase.testDirectoryListing fails on Travis
> -
>
> Key: FLINK-13483
> URL: https://issues.apache.org/jira/browse/FLINK-13483
> Project: Flink
> Issue Type: Bug
> Components: FileSystems
>Reporter: Tzu-Li (Gordon) Tai
>Assignee: Lu Niu
>Priority: Critical
>
> https://api.travis-ci.org/v3/job/564894421/log.txt
> {code}
> PrestoS3FileSystemITCase>AbstractHadoopFileSystemITTest.testDirectoryListing:144->AbstractHadoopFileSystemITTest.checkPathExistence:61
> expected: but was:
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (FLINK-16640) Expose listStatus latency in flink filesystem
Lu Niu created FLINK-16640: -- Summary: Expose listStatus latency in flink filesystem Key: FLINK-16640 URL: https://issues.apache.org/jira/browse/FLINK-16640 Project: Flink Issue Type: Improvement Components: FileSystems Reporter: Lu Niu listStatus could potentially takes long time and slowdown flow workflow. Expose the metrics will help developer better debug. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot edited a comment on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * 1df1ef113620472a0ef36d75899bad699d22438f Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153843153) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6338) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#issuecomment-599596430 ## CI report: * d6eea92060da3afdf541e2fcc90e103c34cef259 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153519796) * acf17c53b9dd8dad8d8e2286c2140f96dca488b3 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153843138) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6337) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-16555) Preflight check for known unstable hashCodes.
[ https://issues.apache.org/jira/browse/FLINK-16555?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061342#comment-17061342 ] Lu Niu commented on FLINK-16555: Is someone working on this? If not, could you assign it to me? I used to work on presto and now move to flink. I think this task would be great to help me understand flink type system. > Preflight check for known unstable hashCodes. > - > > Key: FLINK-16555 > URL: https://issues.apache.org/jira/browse/FLINK-16555 > Project: Flink > Issue Type: Improvement > Components: API / DataStream, API / Type Serialization System >Reporter: Stephan Ewen >Priority: Critical > Labels: usability > Fix For: 1.11.0 > > > Data types can only be used as keys, if they have a stable hash code > implementation that is deterministic across JVMs. Otherwise, the keyBy() > operations will result in incorrect data routing. > We should eagerly check the key type information for known cases types with > unstable hash code, such as > * arrays > * enums > * anything that does not override Object.hashCode -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
flinkbot edited a comment on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#issuecomment-599596430 ## CI report: * d6eea92060da3afdf541e2fcc90e103c34cef259 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153519796) * acf17c53b9dd8dad8d8e2286c2140f96dca488b3 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot commented on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600393951 ## CI report: * 1df1ef113620472a0ef36d75899bad699d22438f UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-13483) PrestoS3FileSystemITCase.testDirectoryListing fails on Travis
[
https://issues.apache.org/jira/browse/FLINK-13483?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061334#comment-17061334
]
Lu Niu commented on FLINK-13483:
Could you assign it to me if no one's working on it? I am a presto contributor
and want to help here.
> PrestoS3FileSystemITCase.testDirectoryListing fails on Travis
> -
>
> Key: FLINK-13483
> URL: https://issues.apache.org/jira/browse/FLINK-13483
> Project: Flink
> Issue Type: Bug
> Components: FileSystems
>Reporter: Tzu-Li (Gordon) Tai
>Priority: Critical
>
> https://api.travis-ci.org/v3/job/564894421/log.txt
> {code}
> PrestoS3FileSystemITCase>AbstractHadoopFileSystemITTest.testDirectoryListing:144->AbstractHadoopFileSystemITTest.checkPathExistence:61
> expected: but was:
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Comment Edited] (FLINK-16517) Add a long running WordCount example
[ https://issues.apache.org/jira/browse/FLINK-16517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061327#comment-17061327 ] Ethan Li edited comment on FLINK-16517 at 3/18/20, 2:26 AM: Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in [readTextFile|#L1085] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. was (Author: ethanli): Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in "[readTextFile|#L1085]] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. > Add a long running WordCount example > > > Key: FLINK-16517 > URL: https://issues.apache.org/jira/browse/FLINK-16517 > Project: Flink > Issue Type: Improvement > Components: Examples >Reporter: Ethan Li >Assignee: Ethan Li >Priority: Minor > Labels: pull-request-available > Time Spent: 10m > Remaining Estimate: 0h > > As far as I know, flink doesn't have a long running WordCount example for > users to start with or doing some simple tests. > The closest one is SocketWindowWordCount. But it requires setting up a server > (nc -l ), which is not hard, but still tedious for simple use cases. And it > requires human input for the job to actually run. > I propose to add or modify current WordCount example to have a SourceFunction > that randomly generates input data based on a set of sentences, so the > WordCount job can run forever. The generation ratio will be configurable. > This will be the easiest way to start a long running flink job and can be > useful for new users to start using flink quickly, or for developers to test > flink easily. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-16637) Flink per job mode terminates before serving job cancellation result
[
https://issues.apache.org/jira/browse/FLINK-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061328#comment-17061328
]
Zili Chen commented on FLINK-16637:
---
[~chaiyq] Thanks for your reminding. I've linked these two issues.
It seems there are two problems here
1. After FLINK-15116 we synchronously shutdown the cluster after job cancelled
so that it can never serve the result.
2. Even without FLINK-15116 there is a risk we don't clear outstanding response
futures before the server down.
Before we agree on these two reason I will keep both of issues open, and I will
try to look into the problem the next day. But tbh, I don't have a plan so far.
> Flink per job mode terminates before serving job cancellation result
>
>
> Key: FLINK-16637
> URL: https://issues.apache.org/jira/browse/FLINK-16637
> Project: Flink
> Issue Type: Bug
> Components: Deployment / YARN
>Affects Versions: 1.10.0
>Reporter: Zili Chen
>Priority: Major
> Fix For: 1.10.1, 1.11.0
>
> Attachments: yarn.log
>
>
> The {{MiniDispatcher}} no longer waits until the REST handler has served the
> cancellation result before shutting down. This behaviour seems to be
> introduced with FLINK-15116.
> See also
> [https://lists.apache.org/x/thread.html/rcadbd6ceede422bac8d4483fd0cdae58659fbff78533a399eb136743@%3Cdev.flink.apache.org%3E]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Comment Edited] (FLINK-16517) Add a long running WordCount example
[ https://issues.apache.org/jira/browse/FLINK-16517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061327#comment-17061327 ] Ethan Li edited comment on FLINK-16517 at 3/18/20, 2:26 AM: Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in [readTextFile|https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java#L1085] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. was (Author: ethanli): Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in [readTextFile|#L1085] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. > Add a long running WordCount example > > > Key: FLINK-16517 > URL: https://issues.apache.org/jira/browse/FLINK-16517 > Project: Flink > Issue Type: Improvement > Components: Examples >Reporter: Ethan Li >Assignee: Ethan Li >Priority: Minor > Labels: pull-request-available > Time Spent: 10m > Remaining Estimate: 0h > > As far as I know, flink doesn't have a long running WordCount example for > users to start with or doing some simple tests. > The closest one is SocketWindowWordCount. But it requires setting up a server > (nc -l ), which is not hard, but still tedious for simple use cases. And it > requires human input for the job to actually run. > I propose to add or modify current WordCount example to have a SourceFunction > that randomly generates input data based on a set of sentences, so the > WordCount job can run forever. The generation ratio will be configurable. > This will be the easiest way to start a long running flink job and can be > useful for new users to start using flink quickly, or for developers to test > flink easily. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Comment Edited] (FLINK-16517) Add a long running WordCount example
[ https://issues.apache.org/jira/browse/FLINK-16517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061327#comment-17061327 ] Ethan Li edited comment on FLINK-16517 at 3/18/20, 2:25 AM: Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in "[readTextFile|#L1085]] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. was (Author: ethanli): Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in "[readTextFile|[https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java#L1085]] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. > Add a long running WordCount example > > > Key: FLINK-16517 > URL: https://issues.apache.org/jira/browse/FLINK-16517 > Project: Flink > Issue Type: Improvement > Components: Examples >Reporter: Ethan Li >Assignee: Ethan Li >Priority: Minor > Labels: pull-request-available > Time Spent: 10m > Remaining Estimate: 0h > > As far as I know, flink doesn't have a long running WordCount example for > users to start with or doing some simple tests. > The closest one is SocketWindowWordCount. But it requires setting up a server > (nc -l ), which is not hard, but still tedious for simple use cases. And it > requires human input for the job to actually run. > I propose to add or modify current WordCount example to have a SourceFunction > that randomly generates input data based on a set of sentences, so the > WordCount job can run forever. The generation ratio will be configurable. > This will be the easiest way to start a long running flink job and can be > useful for new users to start using flink quickly, or for developers to test > flink easily. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-16517) Add a long running WordCount example
[ https://issues.apache.org/jira/browse/FLINK-16517?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061327#comment-17061327 ] Ethan Li commented on FLINK-16517: -- Thanks [~aljoscha] I put up a pull request to add the new unbounded source. I did not deal with FileProcessingMode in this pr because it seems to require some changes in "[readTextFile|[https://github.com/apache/flink/blob/master/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java#L1085]] and that might impact other components/code. I am willing to file a separate issue/PR if you think it makes sense to do so. > Add a long running WordCount example > > > Key: FLINK-16517 > URL: https://issues.apache.org/jira/browse/FLINK-16517 > Project: Flink > Issue Type: Improvement > Components: Examples >Reporter: Ethan Li >Assignee: Ethan Li >Priority: Minor > Labels: pull-request-available > Time Spent: 10m > Remaining Estimate: 0h > > As far as I know, flink doesn't have a long running WordCount example for > users to start with or doing some simple tests. > The closest one is SocketWindowWordCount. But it requires setting up a server > (nc -l ), which is not hard, but still tedious for simple use cases. And it > requires human input for the job to actually run. > I propose to add or modify current WordCount example to have a SourceFunction > that randomly generates input data based on a set of sentences, so the > WordCount job can run forever. The generation ratio will be configurable. > This will be the easiest way to start a long running flink job and can be > useful for new users to start using flink quickly, or for developers to test > flink easily. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] flinkbot commented on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
flinkbot commented on issue #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432#issuecomment-600391497 Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community to review your pull request. We will use this comment to track the progress of the review. ## Automated Checks Last check on commit 1df1ef113620472a0ef36d75899bad699d22438f (Wed Mar 18 02:24:22 UTC 2020) **Warnings:** * No documentation files were touched! Remember to keep the Flink docs up to date! Mention the bot in a comment to re-run the automated checks. ## Review Progress * ❓ 1. The [description] looks good. * ❓ 2. There is [consensus] that the contribution should go into to Flink. * ❓ 3. Needs [attention] from. * ❓ 4. The change fits into the overall [architecture]. * ❓ 5. Overall code [quality] is good. Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process. The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required Bot commands The @flinkbot bot supports the following commands: - `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`) - `@flinkbot approve all` to approve all aspects - `@flinkbot approve-until architecture` to approve everything until `architecture` - `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention - `@flinkbot disapprove architecture` to remove an approval you gave earlier This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-16626) Exception encountered when cancelling a job in yarn per-job mode
[
https://issues.apache.org/jira/browse/FLINK-16626?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061326#comment-17061326
]
chaiyongqiang commented on FLINK-16626:
---
Maybe you'ra right. As far as i see, *flink stop* is going to be removed .Only
*flink cancel* comes into this logic. Also there could be some other situation
i haven't noticed.
> Exception encountered when cancelling a job in yarn per-job mode
>
>
> Key: FLINK-16626
> URL: https://issues.apache.org/jira/browse/FLINK-16626
> Project: Flink
> Issue Type: Bug
> Components: Deployment / YARN
>Affects Versions: 1.10.0
>Reporter: chaiyongqiang
>Priority: Major
>
> When executing the following command to stop a flink job with yarn per-job
> mode, the client keeps retrying untill timeout (1 minutes）and exit with
> failure. But the job stops successfully.
> Command :
> {noformat}
> flink cancel $jobId yid appId
> {noformat}
> The exception on the client side is :
> {quote}bq.
> 2020-03-17 12:32:13,709 DEBUG org.apache.flink.runtime.rest.RestClient -
> Sending request of class class
> org.apache.flink.runtime.rest.messages.EmptyRequestBody to
> ocdt31.aicloud.local:51231/v1/jobs/cc61033484d4c0e7a27a8a2a36f4de7a?mode=cancel
> ...
> 2020-03-17 12:33:11,065 DEBUG org.apache.flink.runtime.rest.RestClient -
> Sending request of class class
> org.apache.flink.runtime.rest.messages.EmptyRequestBody to
> ocdt31.aicloud.local:51231/v1/jobs/cc61033484d4c0e7a27a8a2a36f4de7a?mode=cancel
> 2020-03-17 12:33:14,070 DEBUG org.apache.flink.runtime.rest.RestClient -
> Shutting down rest endpoint.
> 2020-03-17 12:33:14,070 DEBUG org.apache.flink.runtime.rest.RestClient -
> Sending request of class class
> org.apache.flink.runtime.rest.messages.EmptyRequestBody to
> ocdt31.aicloud.local:51231/v1/jobs/cc61033484d4c0e7a27a8a2a36f4de7a?mode=cancel
> 2020-03-17 12:33:14,077 DEBUG
> org.apache.flink.shaded.netty4.io.netty.buffer.PoolThreadCache - Freed 2
> thread-local buffer(s) from thread: flink-rest-client-netty-thread-1
> 2020-03-17 12:33:14,080 DEBUG org.apache.flink.runtime.rest.RestClient - Rest
> endpoint shutdown complete.
> 2020-03-17 12:33:14,083 ERROR org.apache.flink.client.cli.CliFrontend - Error
> while running the command.
> org.apache.flink.util.FlinkException: Could not cancel job
> cc61033484d4c0e7a27a8a2a36f4de7a.
> at
> org.apache.flink.client.cli.CliFrontend.lambda$cancel$7(CliFrontend.java:545)
> at
> org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:843)
> at org.apache.flink.client.cli.CliFrontend.cancel(CliFrontend.java:538)
> at
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:904)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:968)
> at java.security.AccessController.doPrivileged(Native Method)
> at javax.security.auth.Subject.doAs(Subject.java:422)
> at
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1754)
> at
> org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:968)
> Caused by: java.util.concurrent.TimeoutException
> at
> java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1771)
> at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$cancel$7(CliFrontend.java:543)
> ... 9 more
>
> The program finished with the following exception:
> org.apache.flink.util.FlinkException: Could not cancel job
> cc61033484d4c0e7a27a8a2a36f4de7a.
> at
> org.apache.flink.client.cli.CliFrontend.lambda$cancel$7(CliFrontend.java:545)
> at
> org.apache.flink.client.cli.CliFrontend.runClusterAction(CliFrontend.java:843)
> at org.apache.flink.client.cli.CliFrontend.cancel(CliFrontend.java:538)
> at
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:904)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:968)
> at java.security.AccessController.doPrivileged(Native Method)
> at javax.security.auth.Subject.doAs(Subject.java:422)
> at
> org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1754)
> at
> org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:968)
> Caused by: java.util.concurrent.TimeoutException
> at
> java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1771)
> at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)
> at
> org.apache.flink.client.cli.CliF
[jira] [Updated] (FLINK-16517) Add a long running WordCount example
[ https://issues.apache.org/jira/browse/FLINK-16517?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-16517: --- Labels: pull-request-available (was: ) > Add a long running WordCount example > > > Key: FLINK-16517 > URL: https://issues.apache.org/jira/browse/FLINK-16517 > Project: Flink > Issue Type: Improvement > Components: Examples >Reporter: Ethan Li >Assignee: Ethan Li >Priority: Minor > Labels: pull-request-available > > As far as I know, flink doesn't have a long running WordCount example for > users to start with or doing some simple tests. > The closest one is SocketWindowWordCount. But it requires setting up a server > (nc -l ), which is not hard, but still tedious for simple use cases. And it > requires human input for the job to actually run. > I propose to add or modify current WordCount example to have a SourceFunction > that randomly generates input data based on a set of sentences, so the > WordCount job can run forever. The generation ratio will be configurable. > This will be the easiest way to start a long running flink job and can be > useful for new users to start using flink quickly, or for developers to test > flink easily. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] Ethanlm opened a new pull request #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example
Ethanlm opened a new pull request #11432: [FLINK-16517][Examples] Add an embedded input to enable a long-running WordCount example URL: https://github.com/apache/flink/pull/11432 ## What is the purpose of the change This PR modifies the current WordCount example to have a SourceFunction that randomly generates input data based on a set of sentences, so the WordCount job can run forever. The generation interval is configurable. This will be the easiest way to start a long running flink job and can be useful for new users to start using flink quickly, or for developers to test flink easily. ## Brief change log - *Added a new SourceFunction `RandomSentenceSource` that generates random data continuously until cancellation* - *Modified the WordCount example to use the `RandomSentenceSource` as the input optionally* ## Verifying this change This change can be verified as follows: - *Manually verified the change by running a 1 node cluster with 1 JobManagers and 1 TaskManagers, submitted the new WordCount jobs with `--random` option with different intervals and verified the jobs worked correctly.* ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): no - Anything that affects deployment or recovery: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? yes - If yes, how is the feature documented? JavaDocs This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
[ https://issues.apache.org/jira/browse/FLINK-16639?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061325#comment-17061325 ] LakeShen commented on FLINK-16639: -- I am really looking forward to your suggestion .:D If you agree me , I hope I can solve this problem , can you assign this jira for me , thanks. > Flink SQL Kafka source connector, add the no json format filter params when > format.type is json > --- > > Key: FLINK-16639 > URL: https://issues.apache.org/jira/browse/FLINK-16639 > Project: Flink > Issue Type: Improvement > Components: Connectors / Kafka >Reporter: LakeShen >Priority: Minor > Fix For: 1.10.2 > > > In my thought, kafka source connector is the one of most frequently used > connector in flink sql. Flink sql kafka source connector supports the > json,csv or other data format. But there is a problem for json format in > kafka source connector. For example, flink sql kafka source ddl l > like this: > CREATE TABLE team_olap_table ( > a varchar, > b varchar, > ) > with ( > 'connector.type' = 'kafka', > 'connector.version' = '0.10', > 'connector.topic' = topics', > 'connector.properties.0.key' = 'group.id', > 'connector.properties.0.value' = 'hello_world', > 'connector.properties.1.key' = 'bootstrap.servers', > 'connector.properties.1.value' = 'xxx', > 'connector.property-version' = '1', > 'connector.startup-mode' = 'latest-offset', > 'format.type' = 'json', > 'format.property-version' = '1', > 'format.derive-schema' = 'true', > 'update-mode' = 'append' > ); > If the kafka topic messages are not json format ,just one or two records,the > flink sql task will fail-over all the time . > In order to solve this problem , if flink sql source connector use the > json-format, I want to add the 'format.fail-on-not-json-record' param in > flink-json module, if this param is true（default），when read the no-json > records, the flink will fail, if this param is false, the flink sql task will > filter no-json records,the flink task running normally. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Commented] (FLINK-16637) Flink per job mode terminates before serving job cancellation result
[
https://issues.apache.org/jira/browse/FLINK-16637?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061323#comment-17061323
]
chaiyongqiang commented on FLINK-16637:
---
Hi, [~tison], i see this problem too. I created another issue [FLINK-16626
|https://issues.apache.org/jira/browse/FLINK-16626]and give some explantiation
for this issue. Do you have a plan to fix this?
> Flink per job mode terminates before serving job cancellation result
>
>
> Key: FLINK-16637
> URL: https://issues.apache.org/jira/browse/FLINK-16637
> Project: Flink
> Issue Type: Bug
> Components: Deployment / YARN
>Affects Versions: 1.10.0
>Reporter: Zili Chen
>Priority: Major
> Fix For: 1.10.1, 1.11.0
>
> Attachments: yarn.log
>
>
> The {{MiniDispatcher}} no longer waits until the REST handler has served the
> cancellation result before shutting down. This behaviour seems to be
> introduced with FLINK-15116.
> See also
> [https://lists.apache.org/x/thread.html/rcadbd6ceede422bac8d4483fd0cdae58659fbff78533a399eb136743@%3Cdev.flink.apache.org%3E]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Created] (FLINK-16639) Flink SQL Kafka source connector, add the no json format filter params when format.type is json
LakeShen created FLINK-16639: Summary: Flink SQL Kafka source connector, add the no json format filter params when format.type is json Key: FLINK-16639 URL: https://issues.apache.org/jira/browse/FLINK-16639 Project: Flink Issue Type: Improvement Components: Connectors / Kafka Reporter: LakeShen Fix For: 1.10.2 In my thought, kafka source connector is the one of most frequently used connector in flink sql. Flink sql kafka source connector supports the json,csv or other data format. But there is a problem for json format in kafka source connector. For example, flink sql kafka source ddl l like this: CREATE TABLE team_olap_table ( a varchar, b varchar, ) with ( 'connector.type' = 'kafka', 'connector.version' = '0.10', 'connector.topic' = topics', 'connector.properties.0.key' = 'group.id', 'connector.properties.0.value' = 'hello_world', 'connector.properties.1.key' = 'bootstrap.servers', 'connector.properties.1.value' = 'xxx', 'connector.property-version' = '1', 'connector.startup-mode' = 'latest-offset', 'format.type' = 'json', 'format.property-version' = '1', 'format.derive-schema' = 'true', 'update-mode' = 'append' ); If the kafka topic messages are not json format ,just one or two records,the flink sql task will fail-over all the time . In order to solve this problem , if flink sql source connector use the json-format, I want to add the 'format.fail-on-not-json-record' param in flink-json module, if this param is true（default），when read the no-json records, the flink will fail, if this param is false, the flink sql task will filter no-json records,the flink task running normally. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[GitHub] [flink] hequn8128 commented on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
hequn8128 commented on issue #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#issuecomment-600389164 @walterddr Thanks a lot for your nice suggestions. The PR has been updated. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] hequn8128 commented on a change in pull request #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones
hequn8128 commented on a change in pull request #11420: [FLINK-16565][python][ml] Make Pipeline Json compitable between Java and Python if all PipelineStages are Java ones URL: https://github.com/apache/flink/pull/11420#discussion_r394066692 ## File path: flink-python/pyflink/ml/tests/test_pipeline_it_case.py ## @@ -169,3 +169,39 @@ def test_pipeline(self): # the first input is false since 0 + 0 is smaller than the max_sum 14. # the second input is true since 12 + 3 is bigger than the max_sum 14. self.assert_equals(actual, ["false", "true"]) + +def test_pipeline_from_and_to_java_json(self): + +def get_java_pipeline_json(): +wrapper = WrapperTransformer(selected_cols=["a", "b"]) +wrapper._convert_params_to_java(wrapper._j_obj) +j_pipeline = get_gateway().jvm.org.apache.flink.ml.api.core.Pipeline() +j_pipeline.appendStage(wrapper._j_obj) +return j_pipeline.toJson() Review comment: A json string literal sounds good for me. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394065435
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/AbstractTaskManagerFileHandler.java
##
@@ -263,4 +264,10 @@ private void transferFile(ChannelHandlerContext ctx, File
file, HttpRequest http
throw new FlinkException("Could not transfer file " +
file + " to the client.", ioe);
}
}
+
+ protected String getFileName(HandlerRequest
handlerRequest) {
+ return "undefined";
Review comment:
Here use the task manager id and file name as the key to LoadingCache.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Commented] (FLINK-16090) Translate "Table API" page of "Table API & SQL" into Chinese
[
https://issues.apache.org/jira/browse/FLINK-16090?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061307#comment-17061307
]
ShijieZhang commented on FLINK-16090:
-
[~jark], could you assign it to me? I'd like to take this issue.
> Translate "Table API" page of "Table API & SQL" into Chinese
> -
>
> Key: FLINK-16090
> URL: https://issues.apache.org/jira/browse/FLINK-16090
> Project: Flink
> Issue Type: Sub-task
> Components: chinese-translation, Documentation
>Reporter: Jark Wu
>Priority: Major
>
> The page url is
> https://ci.apache.org/projects/flink/flink-docs-master/zh/dev/table/tableApi.html
> The markdown file is located in {{flink/docs/dev/table/tableApi.zh.md}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add log list and read log by name for taskmanager
jinglining commented on a change in pull request #11250: [FLINK-16302][rest]add
log list and read log by name for taskmanager
URL: https://github.com/apache/flink/pull/11250#discussion_r394064136
##
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##
@@ -309,6 +311,25 @@ public TaskExecutor(
return
CompletableFuture.completedFuture(shuffleEnvironment.getPartitionsOccupyingLocalResources().isEmpty());
}
+ @Override
+ public CompletableFuture>>
requestLogList(Time timeout) {
+ final String logDir =
taskManagerConfiguration.getTaskManagerLogDir();
+ if (logDir != null) {
+ final File[] logFiles = new File(logDir).listFiles();
+
+ if (logFiles == null) {
+ return FutureUtils.completedExceptionally(
+ new FlinkException("The specific log
directory is not a valid directory."));
+ }
+
+ final List> logsWithLength =
Arrays.stream(logFiles)
+ .map(logFile -> Tuple2.of(logFile.getName(),
logFile.length()))
+ .collect(Collectors.toList());
+ return
CompletableFuture.completedFuture(logsWithLength);
+ }
+ return FutureUtils.completedExceptionally(new
FlinkException("There is no log file available on the TaskExecutor."));
Review comment:
Because there must be a log file currently, if there is no log file, it
should be an exception, not empty.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[jira] [Assigned] (FLINK-15000) WebUI Metrics is very slow in large parallelism
[
https://issues.apache.org/jira/browse/FLINK-15000?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Jingsong Lee reassigned FLINK-15000:
Assignee: fa zheng
> WebUI Metrics is very slow in large parallelism
> ---
>
> Key: FLINK-15000
> URL: https://issues.apache.org/jira/browse/FLINK-15000
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Web Frontend
>Affects Versions: 1.9.0, 1.9.1
>Reporter: fa zheng
>Assignee: fa zheng
>Priority: Major
> Labels: pull-request-available
> Fix For: 1.11.0
>
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> metrics in web ui are very slow when parallelism is huge. It's hard to add
> metric and choose one metric. I run carTopSpeedWindowingExample with command
> {code:java}
> //代码占位符
> flink run -m yarn-cluster -p 1200 examples/streaming/TopSpeedWindowing.jar
> {code}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-16627) Remove keys with null value in json
[
https://issues.apache.org/jira/browse/FLINK-16627?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061301#comment-17061301
]
Benchao Li commented on FLINK-16627:
[~jark] Sounds reasonable to me, +1 to use 'format.json-include'.
> Remove keys with null value in json
> ---
>
> Key: FLINK-16627
> URL: https://issues.apache.org/jira/browse/FLINK-16627
> Project: Flink
> Issue Type: Improvement
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table
> SQL / Planner
>Affects Versions: 1.10.0
>Reporter: jackray wang
>Assignee: jackray wang
>Priority: Major
>
> {code:java}
> //sql
> CREATE TABLE sink_kafka ( subtype STRING , svt STRING ) WITH (……）
> {code}
>
> {code:java}
> //sql
> CREATE TABLE source_kafka ( subtype STRING , svt STRING ) WITH (……）
> {code}
>
> {code:java}
> //scala udf
> class ScalaUpper extends ScalarFunction {
> def eval(str: String) : String= {
>if(str == null){
>return ""
>}else{
>return str
>}
> }
>
> }
> btenv.registerFunction("scala_upper", new ScalaUpper())
> {code}
>
> {code:java}
> //sql
> insert into sink_kafka select subtype, scala_upper(svt) from source_kafka
> {code}
>
>
>
> Sometimes the svt's value is null, inert into kafkas json like
> \{"subtype":"qin","svt":null}
> If the amount of data is small, it is acceptable，but we process 10TB of data
> every day, and there may be many nulls in the json, which affects the
> efficiency. If you can add a parameter to remove the null key when defining a
> sinktable, the performance will be greatly improved
>
>
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] xintongsong commented on issue #11401: [FLINK-15991][doc][zh] Translate memory configuration docs into Chinese.
xintongsong commented on issue #11401: [FLINK-15991][doc][zh] Translate memory configuration docs into Chinese. URL: https://github.com/apache/flink/pull/11401#issuecomment-600381923 Hi @wuchong, would you have time to take a look at this? This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] PatrickRen commented on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation
PatrickRen commented on issue #11425: [FLINK-16125][kafka] Remove unnecessary zookeeper.connect property validation URL: https://github.com/apache/flink/pull/11425#issuecomment-600378957 Thank you for the review @wuchong ! Sure I'll fix it and create another PR for release-1.10 later. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Issue Comment Deleted] (FLINK-12206) cannot query nested fields using Flink SQL
[
https://issues.apache.org/jira/browse/FLINK-12206?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
LUBIN LIU updated FLINK-12206:
--
Comment: was deleted
(was: We are suffering the same kind of issue. Is this issue solved in the new
version?)
> cannot query nested fields using Flink SQL
> --
>
> Key: FLINK-12206
> URL: https://issues.apache.org/jira/browse/FLINK-12206
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.8.0
>Reporter: Yu Yang
>Priority: Major
>
> We feed list of events with the following RowTypeInfo to flink,
> {code:java}
> Row(
> timestamp: Long,
> userId: Long,
> eventType: String,
> auxData: Map,
> userActions:
> Map,
> diagnostics: Row(hostname: String, ipaddress: String)
> )
> {code}
> and run the following SQL query
> {code:sql}
> SELECT event.userId, event.diagnostics.hostname
> FROM event
> WHERE event.userId < 10;
> {code}
> We are prompted "Column 'diagnostics.hostname' not found in table 'event'".
> Do I miss anything while constructuing the RowTypeInfo? Or it is because any
> SQL validation issue?
> =
> The following is the detailed exceptions:
> org.apache.flink.client.program.ProgramInvocationException: The main method
> caused an error: SQL validation failed. From line 1, column 28 to line 1,
> column 47: Column 'diagnostics.hostname' not found in table 'event'
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)
> at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:423)
> at
> org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:813)
> at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:287)
> at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
> at
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1050)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)
> at
> org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)
> Caused by: org.apache.flink.table.api.ValidationException: SQL validation
> failed. From line 1, column 28 to line 1, column 47: Column
> 'diagnostics.hostname' not found in table 'event'
> at
> org.apache.flink.table.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:109)
> at
> org.apache.flink.table.api.TableEnvironment.sqlQuery(TableEnvironment.scala:746)
> at
> com.pinterest.flink.samples.ThriftRowSerializerSample.main(ThriftRowSerializerSample.java:71)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)
> ... 9 more
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 28 to line 1, column 47: Column 'diagnostics.hostname' not found in
> table 'event'
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
> at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:463)
> at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:783)
> at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:768)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:4764)
> at
> org.apache.calcite.sql.validate.DelegatingScope.fullyQualify(DelegatingScope.java:439)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.visit(SqlValidatorImpl.java:5624)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.visit(SqlValidatorImpl.java:5606)
> at org.apache.calcite.sql.SqlIdentifier.accept(SqlIdentifier.java:334)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.expand(SqlValidatorImpl.java:5213)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.expandSelectItem(SqlValidatorImpl.java:435)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelectList(SqlValidatorImpl.java:4028)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateS
[jira] [Created] (FLINK-16638) Flink checkStateMappingCompleteness doesn't include UserDefinedOperatorIDs
Bashar Abdul Jawad created FLINK-16638:
--
Summary: Flink checkStateMappingCompleteness doesn't include
UserDefinedOperatorIDs
Key: FLINK-16638
URL: https://issues.apache.org/jira/browse/FLINK-16638
Project: Flink
Issue Type: Bug
Components: Runtime / Checkpointing
Affects Versions: 1.10.0
Reporter: Bashar Abdul Jawad
[StateAssignmentOperation.checkStateMappingCompleteness|https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StateAssignmentOperation.java#L555]
doesn't check for UserDefinedOperatorIDs (specified using setUidHash), causing
the exception:
{code}
java.lang.IllegalStateException: There is no operator for the state {}
{code}
to be thrown when a savepoint can't be mapped to an ExecutionJobVertex, even
when the operator hash is explicitly specified.
I believe this logic should be extended to also include UserDefinedOperatorIDs
as so:
{code:java}
for (ExecutionJobVertex executionJobVertex : tasks) {
allOperatorIDs.addAll(executionJobVertex.getOperatorIDs());
allOperatorIDs.addAll(executionJobVertex.getUserDefinedOperatorIDs());
}
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] rkhachatryan commented on issue #11403: [FLINK-16316][operators] Implement new StreamOperatorBase as a replacement for AbstractStreamOperator
rkhachatryan commented on issue #11403: [FLINK-16316][operators] Implement new StreamOperatorBase as a replacement for AbstractStreamOperator URL: https://github.com/apache/flink/pull/11403#issuecomment-600368696 > 1. StreamOperatorStateHandler should be not a field inside StreamOperatorBase, but a wrapper around it, but... > 1. AbstractStreamOperator (and hence I assume StreamOperatorBase as well), needs to expose things from within StreamOperatorStateHandler. > 1. Purpose of AbstractStreamOperator is to provide a convenient class, so that users do not have to implement all of the methods manually. > putting those three together, it's hard for me to imagine a solution, where we split StreamOperatorBase I think an important point is missed from my comment: we should change the **interfaces** first (i.e. `StreamOperator`, not `StreamOperatorBase`). Then much of the need for convenience will go away as well as difficulty of splitting the implementation. > without multiple inheritance it's impossible to provide abstract classes for combination ... I think we can use what is called traits (or mixins) in other languages and can be achieved with default interface methods in Java. > My purpose here was to solve some of the problems that I was aware before (non final transient methods, and incompatibility with MultipleInputStreamOperator interface). To me these are two separate things: 1. providing a base class for `MultipleInputStreamOperator`; for this, if we can't use existing abstraction then why not just copy what we need from `AbstractStreamOperator`? IMO, code duplication in one place is much lesser evil then a wrong abstraction for hundred classes 1. fixing issues in `AbstractStreamOperator`; I totally agree that we need a separate FLIP and a much broader discussion/consideration if we want it to become base class for all operators > Does it mean that you are generally speaking fine with proceeding with the PR as it is? ... I don't think we should replace the `AbstractStreamOperator` now. I'd rather: 1. have a new base class just for the `MultipleInputStreamOperator` case 1. proceed with the refactorings *inside* `AbstractStreamOperator` - but not deprecate it Then, after we get an understanding of how the top-level interface should look and after porting some operators on to new base class we can think of their generalization. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[jira] [Commented] (FLINK-12206) cannot query nested fields using Flink SQL
[
https://issues.apache.org/jira/browse/FLINK-12206?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061271#comment-17061271
]
LIU LUBIN commented on FLINK-12206:
---
We are suffering the same kind of issue. Is this issue solved in the new
version?
> cannot query nested fields using Flink SQL
> --
>
> Key: FLINK-12206
> URL: https://issues.apache.org/jira/browse/FLINK-12206
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
>Affects Versions: 1.8.0
>Reporter: Yu Yang
>Priority: Major
>
> We feed list of events with the following RowTypeInfo to flink,
> {code:java}
> Row(
> timestamp: Long,
> userId: Long,
> eventType: String,
> auxData: Map,
> userActions:
> Map,
> diagnostics: Row(hostname: String, ipaddress: String)
> )
> {code}
> and run the following SQL query
> {code:sql}
> SELECT event.userId, event.diagnostics.hostname
> FROM event
> WHERE event.userId < 10;
> {code}
> We are prompted "Column 'diagnostics.hostname' not found in table 'event'".
> Do I miss anything while constructuing the RowTypeInfo? Or it is because any
> SQL validation issue?
> =
> The following is the detailed exceptions:
> org.apache.flink.client.program.ProgramInvocationException: The main method
> caused an error: SQL validation failed. From line 1, column 28 to line 1,
> column 47: Column 'diagnostics.hostname' not found in table 'event'
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546)
> at
> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)
> at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:423)
> at
> org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:813)
> at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:287)
> at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
> at
> org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1050)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)
> at
> org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)
> Caused by: org.apache.flink.table.api.ValidationException: SQL validation
> failed. From line 1, column 28 to line 1, column 47: Column
> 'diagnostics.hostname' not found in table 'event'
> at
> org.apache.flink.table.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:109)
> at
> org.apache.flink.table.api.TableEnvironment.sqlQuery(TableEnvironment.scala:746)
> at
> com.pinterest.flink.samples.ThriftRowSerializerSample.main(ThriftRowSerializerSample.java:71)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at
> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)
> ... 9 more
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1,
> column 28 to line 1, column 47: Column 'diagnostics.hostname' not found in
> table 'event'
> at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
> at
> sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
> at
> sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
> at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
> at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:463)
> at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:783)
> at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:768)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:4764)
> at
> org.apache.calcite.sql.validate.DelegatingScope.fullyQualify(DelegatingScope.java:439)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.visit(SqlValidatorImpl.java:5624)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl$Expander.visit(SqlValidatorImpl.java:5606)
> at org.apache.calcite.sql.SqlIdentifier.accept(SqlIdentifier.java:334)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.expand(SqlValidatorImpl.java:5213)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.expandSelectItem(SqlValidatorImpl.java:435)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelectList(SqlValidatorImpl.java:4028)
> at
> org.apache.calcite.sql.validate.S
[jira] [Commented] (FLINK-16597) tryCleanupOnError should happen before close
[
https://issues.apache.org/jira/browse/FLINK-16597?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061249#comment-17061249
]
Roman Khachatryan commented on FLINK-16597:
---
Some additional observations:
# `OutputFormatSinkFunction` also firstly closes and then (if `close`
errored) calls `tryCleanupOnError`
# `DataSinkTask#invoke` contains 2 `close` calls (1st one just before catch)
OutputFormatSinkFunction behavior looks right to me because interface
`CleanupWhenUnsuccessful` specifies that it should be called in case of *any*
error including in `close`.
So I'd propose to
# `DataSinkTask#invoke`
## remove 2nd close call
## put 1st one into the new nested finally
## preserve original exception (if close throws itself) - ideally by adding
suppressed
## wrap exception thrown from close (instead of logging)
## don't check if task was cancelled
# `DataSinkTask#cancel`: don't swallow exception; move tryCleanupOnError
inside catch block
> tryCleanupOnError should happen before close
>
>
> Key: FLINK-16597
> URL: https://issues.apache.org/jira/browse/FLINK-16597
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Task
>Affects Versions: 1.10.0
>Reporter: Jiayi Liao
>Priority: Major
>
> If {{DataSinkTask}} fails or throws an exception, the {{OutputFormat}} will
> call {{tryCleanupOnError}} before {{close}}. But the calling order is reverse
> when {{DataSinkTask}} is cancelled, which doesn't make much sense that we can
> still clean the output format after it's closed.
> I'm not very sure if this is a bug. But this does mislead our developers when
> implementing our customized {{OutputFormat}}. cc [~sewen]
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-16620) Add attempt information in logging
[
https://issues.apache.org/jira/browse/FLINK-16620?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061229#comment-17061229
]
Roman Khachatryan commented on FLINK-16620:
---
Thanks for the idea [~wind_ljy].
I think there could be two cases when execution attempt could be useful:
# figuring out what happened with a stream (data) during a particular
execution; for that, we should add it for all log messages and then filter
(which is impractical, unless we implement structured logging)
# figuring out what happened with execution X (e.g. why it failed); for that,
we should add it to start/end
WDYT? Do you have a specific scenario/need in mind?
> Add attempt information in logging
> --
>
> Key: FLINK-16620
> URL: https://issues.apache.org/jira/browse/FLINK-16620
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Task
>Affects Versions: 1.10.0
>Reporter: Jiayi Liao
>Priority: Minor
>
> Currently logging in places such as {{Task}} and {{StreamTask}} , is using
> {{taskNameWithSubtasks}} in {{TaskInfo}} to represent an execution. I think
> it'll be more user-friendly if we can add attempt information into the
> logging.
> The format can be consitent with logging information in {{Execution}} :
> {code:java}
> MySink (3/10) - execution #0
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (FLINK-11852) Improve Processing function example
[
https://issues.apache.org/jira/browse/FLINK-11852?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17061222#comment-17061222
]
Flavio Pompermaier commented on FLINK-11852:
I think the PR is good to merge..anyone taking care of it?
> Improve Processing function example
> ---
>
> Key: FLINK-11852
> URL: https://issues.apache.org/jira/browse/FLINK-11852
> Project: Flink
> Issue Type: Improvement
> Components: API / DataStream, Documentation
>Affects Versions: 1.7.2
>Reporter: Flavio Pompermaier
>Assignee: TANG Wen-hui
>Priority: Minor
> Labels: pull-request-available
> Time Spent: 10m
> Remaining Estimate: 0h
>
> In the processing function documentation
> ([https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html)]
> there's an "abusive" usage of the timers since a new timer is registered for
> every new tuple coming in. This could cause problems in terms of allocated
> objects and could burden the overall application.
> It could worth to mention this problem and remove useless timers, e.g.:
>
> {code:java}
> CountWithTimestamp current = state.value();
> if (current == null) {
> current = new CountWithTimestamp();
> current.key = value.f0;
> } else {
> ctx.timerService().deleteEventTimeTimer(current.lastModified + timeout);
> }{code}
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [SUCCESS](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #61: [FLINK-16124] [kinesis] Support Java Kinesis Ingress / Egress
igalshilman commented on a change in pull request #61: [FLINK-16124] [kinesis]
Support Java Kinesis Ingress / Egress
URL: https://github.com/apache/flink-statefun/pull/61#discussion_r393980552
##
File path:
statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/AwsAuthConfig.java
##
@@ -0,0 +1,128 @@
+/*
+ * 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.flink.statefun.flink.io.kinesis;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.Properties;
+import
org.apache.flink.kinesis.shaded.com.amazonaws.regions.DefaultAwsRegionProviderChain;
+import org.apache.flink.statefun.sdk.kinesis.auth.AwsCredentials;
+import org.apache.flink.statefun.sdk.kinesis.auth.AwsRegion;
+import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;
+
+final class AwsAuthConfig {
+ private final AwsRegion awsRegion;
+ private final AwsCredentials awsCredentials;
+
+ AwsAuthConfig(AwsRegion awsRegion, AwsCredentials awsCredentials) {
+this.awsRegion = Objects.requireNonNull(awsRegion);
+this.awsCredentials = Objects.requireNonNull(awsCredentials);
+ }
+
+ Properties asFlinkConnectorProperties() {
+final Properties properties = new Properties();
+setAwsRegionProperties(properties, awsRegion);
+setAwsCredentialProperties(properties, awsCredentials);
+
+return properties;
+ }
+
+ private static void setAwsRegionProperties(Properties properties, AwsRegion
awsRegion) {
+if (awsRegion.isDefault()) {
+ properties.setProperty(AWSConfigConstants.AWS_REGION,
regionFromDefaultProviderChain());
+} else if (awsRegion.isId()) {
+ properties.setProperty(AWSConfigConstants.AWS_REGION,
awsRegion.asId().id());
+} else if (awsRegion.isCustomEndpoint()) {
+ final AwsRegion.CustomEndpointAwsRegion customEndpoint =
awsRegion.asCustomEndpoint();
+ setCustomEndpointForConsumer(properties, customEndpoint);
+ setCustomEndpointForProducer(properties, customEndpoint);
+} else {
+ throw new IllegalStateException("Unrecognized AWS region configuration
type: " + awsRegion);
+}
+ }
+
+ private static void setAwsCredentialProperties(
+ Properties properties, AwsCredentials awsCredentials) {
+if (awsCredentials.isDefault()) {
+ properties.setProperty(
+ AWSConfigConstants.AWS_CREDENTIALS_PROVIDER,
+ AWSConfigConstants.CredentialProvider.AUTO.name());
+} else if (awsCredentials.isBasic()) {
+ properties.setProperty(
+ AWSConfigConstants.AWS_CREDENTIALS_PROVIDER,
+ AWSConfigConstants.CredentialProvider.BASIC.name());
+
+ final AwsCredentials.BasicAwsCredentials basicCredentials =
awsCredentials.asBasic();
+ properties.setProperty(
+
AWSConfigConstants.accessKeyId(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER),
+ basicCredentials.accessKeyId());
+ properties.setProperty(
+
AWSConfigConstants.secretKey(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER),
+ basicCredentials.secretAccessKey());
+} else if (awsCredentials.isProfile()) {
+ properties.setProperty(
+ AWSConfigConstants.AWS_CREDENTIALS_PROVIDER,
+ AWSConfigConstants.CredentialProvider.PROFILE.name());
+
+ final AwsCredentials.ProfileAwsCredentials profileCredentials =
awsCredentials.asProfile();
+ properties.setProperty(
+
AWSConfigConstants.profileName(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER),
+ profileCredentials.name());
+ profileCredentials
+ .path()
+ .ifPresent(
+ path ->
+ properties.setProperty(
+
AWSConfigConstants.profilePath(AWSConfigConstants.AWS_CREDENTIALS_PROVIDER),
+ path));
+} else {
+ throw new IllegalStateException(
+ "Unrecognized AWS credentials configuration type: " +
awsCredentials);
+}
+ }
+
+ private static String regionFromDefaultProviderChain() {
+return new
DefaultAwsRegionProviderChain().getRegion().toLowerCase(Locale.ENGLISH);
+ }
+
+ private static void setCust
[GitHub] [flink-statefun] igalshilman commented on a change in pull request #61: [FLINK-16124] [kinesis] Support Java Kinesis Ingress / Egress
igalshilman commented on a change in pull request #61: [FLINK-16124] [kinesis]
Support Java Kinesis Ingress / Egress
URL: https://github.com/apache/flink-statefun/pull/61#discussion_r393980916
##
File path:
statefun-flink/statefun-flink-io-bundle/src/main/java/org/apache/flink/statefun/flink/io/kinesis/AwsAuthConfig.java
##
@@ -0,0 +1,128 @@
+/*
+ * 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.flink.statefun.flink.io.kinesis;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.Properties;
+import
org.apache.flink.kinesis.shaded.com.amazonaws.regions.DefaultAwsRegionProviderChain;
+import org.apache.flink.statefun.sdk.kinesis.auth.AwsCredentials;
+import org.apache.flink.statefun.sdk.kinesis.auth.AwsRegion;
+import org.apache.flink.streaming.connectors.kinesis.config.AWSConfigConstants;
+
+final class AwsAuthConfig {
+ private final AwsRegion awsRegion;
+ private final AwsCredentials awsCredentials;
+
+ AwsAuthConfig(AwsRegion awsRegion, AwsCredentials awsCredentials) {
+this.awsRegion = Objects.requireNonNull(awsRegion);
+this.awsCredentials = Objects.requireNonNull(awsCredentials);
+ }
+
+ Properties asFlinkConnectorProperties() {
+final Properties properties = new Properties();
+setAwsRegionProperties(properties, awsRegion);
+setAwsCredentialProperties(properties, awsCredentials);
+
+return properties;
+ }
+
+ private static void setAwsRegionProperties(Properties properties, AwsRegion
awsRegion) {
+if (awsRegion.isDefault()) {
+ properties.setProperty(AWSConfigConstants.AWS_REGION,
regionFromDefaultProviderChain());
+} else if (awsRegion.isId()) {
+ properties.setProperty(AWSConfigConstants.AWS_REGION,
awsRegion.asId().id());
+} else if (awsRegion.isCustomEndpoint()) {
+ final AwsRegion.CustomEndpointAwsRegion customEndpoint =
awsRegion.asCustomEndpoint();
+ setCustomEndpointForConsumer(properties, customEndpoint);
+ setCustomEndpointForProducer(properties, customEndpoint);
+} else {
+ throw new IllegalStateException("Unrecognized AWS region configuration
type: " + awsRegion);
+}
+ }
+
+ private static void setAwsCredentialProperties(
Review comment:
This could be a nice method to test.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067) * f992a52b047e8f44480f84bfa398007e2a649bb4 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/153812595) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6335) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11195: [FLINK-16222][runtime] Use plugins mechanism for initializing MetricReporters
flinkbot edited a comment on issue #11195: [FLINK-16222][runtime] Use plugins mechanism for initializing MetricReporters URL: https://github.com/apache/flink/pull/11195#issuecomment-590065588 ## CI report: * 738a6fd162d5ba99432438d224f3b4890961af56 UNKNOWN * 77f24b292a04fde34f4c645151484cf86a13832c UNKNOWN * 5dfb306ba6ff07f700b1afc21847641321705d4d UNKNOWN * 5b0f219689e81fe11c3511c13c7f4943a997a4ef UNKNOWN * a244cee274bab74683c25f90bfd515698bc04c95 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/153812547) Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog
flinkbot edited a comment on issue #11336: [FLINK-16471][jdbc] develop PostgresCatalog URL: https://github.com/apache/flink/pull/11336#issuecomment-595991436 ## CI report: * 7825f1a2049defb7c15de4f2fb688799c108d01e Travis: [SUCCESS](https://travis-ci.com/flink-ci/flink/builds/152384622) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=6067) * f992a52b047e8f44480f84bfa398007e2a649bb4 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@flinkbot run travis` re-run the last Travis build - `@flinkbot run azure` re-run the last Azure build This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
