[jira] [Closed] (FLINK-30220) Secure user credentials in Flink SQL JDBC connector
[
https://issues.apache.org/jira/browse/FLINK-30220?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser closed FLINK-30220.
--
Resolution: Duplicate
[~qinjunjerry] I've closed it since the ticket you're linking to already talks
about a credential mechanism for Flink SQL in general.
> Secure user credentials in Flink SQL JDBC connector
> ---
>
> Key: FLINK-30220
> URL: https://issues.apache.org/jira/browse/FLINK-30220
> Project: Flink
> Issue Type: Improvement
>Reporter: Jun Qin
>Priority: Major
>
> Similar to FLINK-28028, when using Flink SQL JDBC connector, we should also
> have a way to secure the username and the password used in the DDL:
> {code:java}
> CREATE TABLE MyUserTable (
> id BIGINT,
> name STRING,
> age INT,
> status BOOLEAN,
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
>'connector' = 'jdbc',
>'url' = 'jdbc:mysql://localhost:3306/mydatabase',
>'table-name' = 'users',
>'username' = 'a-username',
>'password' = 'a-password'
> );
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[GitHub] [flink] Zakelly commented on a diff in pull request #21362: [FLINK-29430] Add sanity check when setCurrentKeyGroupIndex
Zakelly commented on code in PR #21362:
URL: https://github.com/apache/flink/pull/21362#discussion_r1034413200
##
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/InternalKeyContextImpl.java:
##
@@ -72,6 +73,10 @@ public void setCurrentKey(@Nonnull K currentKey) {
@Override
public void setCurrentKeyGroupIndex(int currentKeyGroupIndex) {
+if (!keyGroupRange.contains(currentKeyGroupIndex)) {
+throw KeyGroupRangeOffsets.newIllegalKeyGroupException(
+currentKeyGroupIndex, keyGroupRange);
+}
Review Comment:
> How about the opposite: removing the check from HashMapStateBackend
(StateTable) and relying on `InternalKeyContext`? So that it works for any
backend. As a side benefit, that would be a bit faster because the check would
be done once per input record, not per state access.
I Agree. That's what I'm trying to say.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[jira] [Commented] (FLINK-30159) Add Transformer for ANOVATest
[ https://issues.apache.org/jira/browse/FLINK-30159?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640450#comment-17640450 ] Jiang Xin commented on FLINK-30159: --- master: d6a5b42aae2a1a84789dfe0a35c7e97461b5d418 > Add Transformer for ANOVATest > - > > Key: FLINK-30159 > URL: https://issues.apache.org/jira/browse/FLINK-30159 > Project: Flink > Issue Type: Sub-task > Components: Library / Machine Learning >Reporter: Jiang Xin >Priority: Major > Labels: pull-request-available > Fix For: ml-2.2.0 > > > Add Transformer for ANOVATest. > Its function would be at least equivalent to Spark's > org.apache.spark.ml.stat.ANOVATest. The relevant PR should contain the > following components: > * Java implementation/test (Must include) > * Python implementation/test -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Resolved] (FLINK-30159) Add Transformer for ANOVATest
[ https://issues.apache.org/jira/browse/FLINK-30159?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Jiang Xin resolved FLINK-30159. --- Resolution: Fixed > Add Transformer for ANOVATest > - > > Key: FLINK-30159 > URL: https://issues.apache.org/jira/browse/FLINK-30159 > Project: Flink > Issue Type: Sub-task > Components: Library / Machine Learning >Reporter: Jiang Xin >Priority: Major > Labels: pull-request-available > Fix For: ml-2.2.0 > > > Add Transformer for ANOVATest. > Its function would be at least equivalent to Spark's > org.apache.spark.ml.stat.ANOVATest. The relevant PR should contain the > following components: > * Java implementation/test (Must include) > * Python implementation/test -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] lsyldliu commented on pull request #21401: [FLINK-29718][table] Supports hive sum function by native implementation
lsyldliu commented on PR #21401: URL: https://github.com/apache/flink/pull/21401#issuecomment-1330242565 @flinkbot run azure -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-30070) Create savepoints without side effects
[
https://issues.apache.org/jira/browse/FLINK-30070?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Piotr Nowojski updated FLINK-30070:
---
Description:
Side effects are any external state - a state that is stored not in Flink, but
in an external system, like for example connectors transactions (KafkaSink,
...).
We shouldn't be relaying on the external systems for storing part of the job's
state, especially for any long period of time. The most prominent issue is that
Kafka transactions can time out, leading to a data loss if transaction hasn't
been committed.
Stop-with-savepoint, currently guarantee that {{notifyCheckpointCompleted}}
call will be issued, so properly implemented operators are guaranteed to
committed it's state. However this information is currently not stored in the
checkpoint in any way ( FLINK-16419 ). Larger issue is how to deal with
savepoints, since there we currently do not have any guarantees that
transactions have been committed.
Some potential solution might be to expand API (like {{CheckpointedFunction}}
), to let the operators/functions know, that they should
close/commit/clear/deal with external state differently and use that API during
stop-with-savepoint and intermediate savepoints. Note that since Flink 1.15,
intermediate savepoints are never committed, so most likely they shouldn't even
try to store/pre-commit any external state/transactions.
was:
Side effects are any external state - a state that is stored not in Flink, but
in an external system, like for example connectors transactions (KafkaSink,
...).
We shouldn't be relaying on the external systems for storing part of the job's
state, especially for any long period of time. The most prominent issue is that
Kafka transactions can time out, leading to a data loss if transaction hasn't
been committed.
Stop-with-savepoint, currently guarantee that {{notifyCheckpointCompleted}}
call will be issued, so properly implemented operators are guaranteed to
committed it's state. However this information is currently not stored in the
checkpoint in any way ( FLINK-16419 ). Larger issue is how to deal with
savepoints, since there we currently do not have any guarantees that
transactions have been committed.
Some potential solution might be to expand API (like {{CheckpointedFunction}}
), to let the operators/functions know, that they should
close/commit/clear/deal with external state differently and use that API during
stop-with-savepoint + rework how regular savepoints are handled.
> Create savepoints without side effects
> --
>
> Key: FLINK-30070
> URL: https://issues.apache.org/jira/browse/FLINK-30070
> Project: Flink
> Issue Type: New Feature
> Components: API / DataStream, Runtime / Checkpointing
>Affects Versions: 1.16.0, 1.15.2, 1.14.6
>Reporter: Piotr Nowojski
>Priority: Major
>
> Side effects are any external state - a state that is stored not in Flink,
> but in an external system, like for example connectors transactions
> (KafkaSink, ...).
> We shouldn't be relaying on the external systems for storing part of the
> job's state, especially for any long period of time. The most prominent issue
> is that Kafka transactions can time out, leading to a data loss if
> transaction hasn't been committed.
> Stop-with-savepoint, currently guarantee that {{notifyCheckpointCompleted}}
> call will be issued, so properly implemented operators are guaranteed to
> committed it's state. However this information is currently not stored in the
> checkpoint in any way ( FLINK-16419 ). Larger issue is how to deal with
> savepoints, since there we currently do not have any guarantees that
> transactions have been committed.
> Some potential solution might be to expand API (like {{CheckpointedFunction}}
> ), to let the operators/functions know, that they should
> close/commit/clear/deal with external state differently and use that API
> during stop-with-savepoint and intermediate savepoints. Note that since Flink
> 1.15, intermediate savepoints are never committed, so most likely they
> shouldn't even try to store/pre-commit any external state/transactions.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-29856) Triggering savepoint does not trigger operator notifyCheckpointComplete
[
https://issues.apache.org/jira/browse/FLINK-29856?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640461#comment-17640461
]
Piotr Nowojski commented on FLINK-29856:
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
That's not entirely true. Imagine a scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uid}}s
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
{{quote}}
one wants to be safe in those scenarios, we advise dropping the state of
transactional sinks, by changing sinks uids.
{{quote}}
Generally speaking recovery to savepoints is always tricky, as in many cases it
can violate exactly-once guarantees. Users have to take that into account, or
use only stop-with-savepoint, to make sure that the original job won't be able
to make any progress while we are doing something with that savepoint.
{{quote}}
Without notifyCheckpointComplete, it isn't clear what the benefit of an
intermediate Savepoint is (need to re-read the FLIP). From the user
perspective, trigger intermediate Savepoint has the benefit to be able to
commit data to external systems like Kafka/Iceberg on demand for their
operational procedures. Perhaps, the eventual solution is to replace
operational procedure with triggering checkpoint with
https://issues.apache.org/jira/browse/FLINK-29634 since that would match the
effects (notifyCheckpointComplete, etc).
{{quote}}
[~mason6345], can you explain what's the actual problem that you are
experiencing and why are you using intermediate savepoints? It sounds to me
like you are using it not the way they were intended. On demand committing data
to external systems sounds like a use case for FLINK-27101.
> Triggering savepoint does not trigger operator notifyCheckpointComplete
> ---
>
> Key: FLINK-29856
> URL: https://issues.apache.org/jira/browse/FLINK-29856
> Project: Flink
> Issue Type: Improvement
> Components: Runtime / Checkpointing
>Affects Versions: 1.15.0, 1.16.0
>Reporter: Mason Chen
>Priority: Major
>
> When I trigger a savepoint with the Flink K8s operator, I verified for two
> sources (KafkaSource and MultiClusterKafkaSource) do not invoke
> notifyCheckpointComplete. This is easily reproducible in a simple pipeline
> (e.g. KafkaSource -> print). In this case, the savepoint is complete and
> successful, which is verified by the Flink Checkpoint UI tab and the
> jobmanager logs. e.g. `
> Triggering checkpoint 3 (type=SavepointType\{name='Savepoint',
> postCheckpointAction=NONE, formatType=CANONICAL})`
>
> However, when the checkpoint occurs via the interval, I do see the sources
> checkpointing properly and expected logs in the output.
> After the ticket was initially filed, I also checked with other stateful UDFs
> and observed the same behavior.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-29856) Triggering savepoint does not trigger operator notifyCheckpointComplete
[
https://issues.apache.org/jira/browse/FLINK-29856?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640461#comment-17640461
]
Piotr Nowojski edited comment on FLINK-29856 at 11/29/22 8:29 AM:
--
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
That's not entirely true. Imagine a scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uid}}s
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
{quote}
one wants to be safe in those scenarios, we advise dropping the state of
transactional sinks, by changing sinks uids.
{quote}
Generally speaking recovery to savepoints is always tricky, as in many cases it
can violate exactly-once guarantees. Users have to take that into account, or
use only stop-with-savepoint, to make sure that the original job won't be able
to make any progress while we are doing something with that savepoint.
{quote}
Without notifyCheckpointComplete, it isn't clear what the benefit of an
intermediate Savepoint is (need to re-read the FLIP). From the user
perspective, trigger intermediate Savepoint has the benefit to be able to
commit data to external systems like Kafka/Iceberg on demand for their
operational procedures. Perhaps, the eventual solution is to replace
operational procedure with triggering checkpoint with
https://issues.apache.org/jira/browse/FLINK-29634 since that would match the
effects (notifyCheckpointComplete, etc).
{quote}
[~mason6345], can you explain what's the actual problem that you are
experiencing and why are you using intermediate savepoints? It sounds to me
like you are using it not the way they were intended. On demand committing data
to external systems sounds like a use case for FLINK-27101.
was (Author: pnowojski):
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
That's not entirely true. Imagine a scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uid}}s
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
{{quote}}
one wants to be safe in those scenarios
[jira] [Comment Edited] (FLINK-29856) Triggering savepoint does not trigger operator notifyCheckpointComplete
[
https://issues.apache.org/jira/browse/FLINK-29856?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640461#comment-17640461
]
Piotr Nowojski edited comment on FLINK-29856 at 11/29/22 8:31 AM:
--
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
The real rationale behind the current behaviour is if you imagine this scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uids}}
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
{quote}
one wants to be safe in those scenarios, we advise dropping the state of
transactional sinks, by changing sinks uids.
{quote}
Generally speaking recovery to savepoints is always tricky, as in many cases it
can violate exactly-once guarantees. Users have to take that into account, or
use only stop-with-savepoint, to make sure that the original job won't be able
to make any progress while we are doing something with that savepoint.
{quote}
Without notifyCheckpointComplete, it isn't clear what the benefit of an
intermediate Savepoint is (need to re-read the FLIP). From the user
perspective, trigger intermediate Savepoint has the benefit to be able to
commit data to external systems like Kafka/Iceberg on demand for their
operational procedures. Perhaps, the eventual solution is to replace
operational procedure with triggering checkpoint with
https://issues.apache.org/jira/browse/FLINK-29634 since that would match the
effects (notifyCheckpointComplete, etc).
{quote}
[~mason6345], can you explain what's the actual problem that you are
experiencing and why are you using intermediate savepoints? It sounds to me
like you are using it not the way they were intended. On demand committing data
to external systems sounds like a use case for FLINK-27101.
was (Author: pnowojski):
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
The real rationale behind the current behaviour is if you imagine this scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uid}}s
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#trigg
[jira] [Comment Edited] (FLINK-29856) Triggering savepoint does not trigger operator notifyCheckpointComplete
[
https://issues.apache.org/jira/browse/FLINK-29856?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640461#comment-17640461
]
Piotr Nowojski edited comment on FLINK-29856 at 11/29/22 8:31 AM:
--
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
The real rationale behind the current behaviour is if you imagine this scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uid}}s
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
{quote}
one wants to be safe in those scenarios, we advise dropping the state of
transactional sinks, by changing sinks uids.
{quote}
Generally speaking recovery to savepoints is always tricky, as in many cases it
can violate exactly-once guarantees. Users have to take that into account, or
use only stop-with-savepoint, to make sure that the original job won't be able
to make any progress while we are doing something with that savepoint.
{quote}
Without notifyCheckpointComplete, it isn't clear what the benefit of an
intermediate Savepoint is (need to re-read the FLIP). From the user
perspective, trigger intermediate Savepoint has the benefit to be able to
commit data to external systems like Kafka/Iceberg on demand for their
operational procedures. Perhaps, the eventual solution is to replace
operational procedure with triggering checkpoint with
https://issues.apache.org/jira/browse/FLINK-29634 since that would match the
effects (notifyCheckpointComplete, etc).
{quote}
[~mason6345], can you explain what's the actual problem that you are
experiencing and why are you using intermediate savepoints? It sounds to me
like you are using it not the way they were intended. On demand committing data
to external systems sounds like a use case for FLINK-27101.
was (Author: pnowojski):
Yes, this is the intended behavour.
{quote}
I think the reason why Flink doesn't call notifyCheckpointComplete() anymore on
"intermediate" savepoints is purely for recovery reasons where we want to
ensure that we only call notifyCheckpointComplete() once. Arguably, this isn't
really the case because we can fail directly after restoring from a checkpoint
but at least we will then only commit the already committed data and not any
older data, as would be the case when a savepoint had been committing the data
before we fell back to an earlier checkpoint.
{quote}
That's not entirely true. Imagine a scenario:
# triggering and completing {{chk42}}
# triggering and completing savepoint
# job failover
Job can not recover to the savepoint, because there are no guarantees that it
still exists (savepoint are always owned by the user). It has to failover to
the {{chk42}}. However if we had committed the savepoint, that would have meant
recovery to {{chk42}} would create either duplicate results (if the job is
deterministic) OR inconsistent result (if the job is non-deterministic).
[~mason6345], think about the savepoints as either state backups of your job,
or terminal states after stopping the job (also kind of backups that out live
your jobs). Ideally, neither of those should contain any pointers to an
external state (like Kafka transactions), but this is not yet implemented
(FLINK-30070). As it is, during recovery from savepoints it's recommended to
drop the external state via replacing transactional operators {{uid}}s
(https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/savepoints/#triggering-savepoints).
{quote}
one wa
[jira] [Commented] (FLINK-30236) UpsertKafkaTableITCase failed due to timeout when creating a topic
[
https://issues.apache.org/jira/browse/FLINK-30236?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640464#comment-17640464
]
Gabor Somogyi commented on FLINK-30236:
---
Topic creation can fail because of the Kafka broker. This is hard to find out
because such cases we need broker logs which is running in a different
container. Double pain how to debug deadlock from logs.
The test most probably is passing because Kafka broker auto topic creation is
enabled by default (auto.create.topics.enable). When it's enabled then manual
topic creation is just additional code because any producer can create
arbitrary topics :)
> UpsertKafkaTableITCase failed due to timeout when creating a topic
> --
>
> Key: FLINK-30236
> URL: https://issues.apache.org/jira/browse/FLINK-30236
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
>Affects Versions: 1.17.0
>Reporter: Matthias Pohl
>Priority: Major
> Labels: test-stability
>
> The following build failed due to a timeout when creating the topic.
> {code:java}
> Nov 28 16:25:44 [ERROR] Tests run: 15, Failures: 0, Errors: 1, Skipped: 0,
> Time elapsed: 72.815 s <<< FAILURE! - in
> org.apache.flink.streaming.connectors.kafka.table.UpsertKafkaTableITCase
> Nov 28 16:25:44 [ERROR] UpsertKafkaTableITCase.testAggregate Time elapsed:
> 31.629 s <<< ERROR!
> Nov 28 16:25:44 java.lang.IllegalStateException: Fail to create topic
> [word_count_json partitions: 4 replication factor: 1].
> Nov 28 16:25:44 at
> org.apache.flink.streaming.connectors.kafka.table.KafkaTableTestBase.createTestTopic(KafkaTableTestBase.java:143)
> Nov 28 16:25:44 at
> org.apache.flink.streaming.connectors.kafka.table.UpsertKafkaTableITCase.testAggregate(UpsertKafkaTableITCase.java:78)
> Nov 28 16:25:44 at sun.reflect.NativeMethodAccessorImpl.invoke0(Native
> Method)
> Nov 28 16:25:44 at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> [...] {code}
> I suspect this being a general issue with providing timeouts for tests.
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=43566&view=logs&j=aa18c3f6-13b8-5f58-86bb-c1cffb239496&t=502fb6c0-30a2-5e49-c5c2-a00fa3acb203&l=37623
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[GitHub] [flink] MartijnVisser commented on pull request #21279: Bump socket.io-parser from 4.0.4 to 4.0.5 in /flink-runtime-web/web-dashboard
MartijnVisser commented on PR #21279: URL: https://github.com/apache/flink/pull/21279#issuecomment-1330269796 @flinkbot run azure -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (FLINK-29634) Support periodic checkpoint triggering
[ https://issues.apache.org/jira/browse/FLINK-29634?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640466#comment-17640466 ] Piotr Nowojski commented on FLINK-29634: I'm not sure if Flink should try to satisfy more and more non-standard checkpoint triggering behaviours for small fraction of users. One of the motivations behind FLINK-27101 was to allow users to implement their own logic outside of Flink, if they really need it. > Support periodic checkpoint triggering > -- > > Key: FLINK-29634 > URL: https://issues.apache.org/jira/browse/FLINK-29634 > Project: Flink > Issue Type: New Feature > Components: Kubernetes Operator >Reporter: Thomas Weise >Assignee: Jiale Tan >Priority: Major > > Similar to the support for periodic savepoints, the operator should support > triggering periodic checkpoints to break the incremental checkpoint chain. > Support for external triggering will come with 1.17: > https://issues.apache.org/jira/browse/FLINK-27101 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-27916) HybridSourceReaderTest.testReader failed with AssertionError
[
https://issues.apache.org/jira/browse/FLINK-27916?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640467#comment-17640467
]
Matthias Pohl commented on FLINK-27916:
---
[~thomasWeise]/[~binh] may you have a look at it considering that it happened a
few times recently again?
> HybridSourceReaderTest.testReader failed with AssertionError
>
>
> Key: FLINK-27916
> URL: https://issues.apache.org/jira/browse/FLINK-27916
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Common
>Affects Versions: 1.16.0
>Reporter: Huang Xingbo
>Priority: Major
> Labels: auto-deprioritized-critical, test-stability
> Attachments: Screen Shot 2022-07-21 at 5.51.40 PM.png
>
>
> {code:java}
> 2022-06-05T07:47:33.3332158Z Jun 05 07:47:33 [ERROR] Tests run: 3, Failures:
> 1, Errors: 0, Skipped: 0, Time elapsed: 2.03 s <<< FAILURE! - in
> org.apache.flink.connector.base.source.hybrid.HybridSourceReaderTest
> 2022-06-05T07:47:33.3334366Z Jun 05 07:47:33 [ERROR]
> org.apache.flink.connector.base.source.hybrid.HybridSourceReaderTest.testReader
> Time elapsed: 0.108 s <<< FAILURE!
> 2022-06-05T07:47:33.3335385Z Jun 05 07:47:33 java.lang.AssertionError:
> 2022-06-05T07:47:33.3336049Z Jun 05 07:47:33
> 2022-06-05T07:47:33.3336682Z Jun 05 07:47:33 Expected size: 1 but was: 0 in:
> 2022-06-05T07:47:33.3337316Z Jun 05 07:47:33 []
> 2022-06-05T07:47:33.3338437Z Jun 05 07:47:33 at
> org.apache.flink.connector.base.source.hybrid.HybridSourceReaderTest.assertAndClearSourceReaderFinishedEvent(HybridSourceReaderTest.java:199)
> 2022-06-05T07:47:33.3340082Z Jun 05 07:47:33 at
> org.apache.flink.connector.base.source.hybrid.HybridSourceReaderTest.testReader(HybridSourceReaderTest.java:96)
> 2022-06-05T07:47:33.3341373Z Jun 05 07:47:33 at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 2022-06-05T07:47:33.3342540Z Jun 05 07:47:33 at
> java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 2022-06-05T07:47:33.3344124Z Jun 05 07:47:33 at
> java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 2022-06-05T07:47:33.3345283Z Jun 05 07:47:33 at
> java.base/java.lang.reflect.Method.invoke(Method.java:566)
> 2022-06-05T07:47:33.3346804Z Jun 05 07:47:33 at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
> 2022-06-05T07:47:33.3348218Z Jun 05 07:47:33 at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> 2022-06-05T07:47:33.3349495Z Jun 05 07:47:33 at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
> 2022-06-05T07:47:33.3350779Z Jun 05 07:47:33 at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> 2022-06-05T07:47:33.3351956Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
> 2022-06-05T07:47:33.3357032Z Jun 05 07:47:33 at
> org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
> 2022-06-05T07:47:33.3358633Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
> 2022-06-05T07:47:33.3360003Z Jun 05 07:47:33 at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
> 2022-06-05T07:47:33.3361924Z Jun 05 07:47:33 at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
> 2022-06-05T07:47:33.3363427Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
> 2022-06-05T07:47:33.3364793Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
> 2022-06-05T07:47:33.3365619Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
> 2022-06-05T07:47:33.3366254Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
> 2022-06-05T07:47:33.3366939Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
> 2022-06-05T07:47:33.3367556Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
> 2022-06-05T07:47:33.3368268Z Jun 05 07:47:33 at
> org.junit.runners.ParentRunner.run(ParentRunner.java:413)
> 2022-06-05T07:47:33.3369166Z Jun 05 07:47:33 at
> org.junit.runner.JUnitCore.run(JUnitCore.java:137)
> 2022-06-05T07:47:33.3369993Z Jun 05 07:47:33 at
> org.junit.runner.JUnitCore.run(JUnitCore.java:115)
> 2022-06-05T07:47:33.3371021Z Jun 05 07:47:33 at
> org.junit.vintage.engine.execution.RunnerExecutor.execute(RunnerExecutor.java:42)
> 2022-06-05T07:47:33.3372128Z Jun 05 07:47:33 at
> org.junit.vintage.engine.VintageTestEngine.executeAllChildren(VintageTe
[jira] [Commented] (FLINK-18356) flink-table-planner Exit code 137 returned from process
[
https://issues.apache.org/jira/browse/FLINK-18356?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640471#comment-17640471
]
Matthias Pohl commented on FLINK-18356:
---
[~godfrey] Could you have a look at the OOM error? It appears quite frequently
recently.
> flink-table-planner Exit code 137 returned from process
> ---
>
> Key: FLINK-18356
> URL: https://issues.apache.org/jira/browse/FLINK-18356
> Project: Flink
> Issue Type: Bug
> Components: Build System / Azure Pipelines, Tests
>Affects Versions: 1.12.0, 1.13.0, 1.14.0, 1.15.0, 1.16.0, 1.17.0
>Reporter: Piotr Nowojski
>Priority: Critical
> Labels: pull-request-available, test-stability
> Attachments: 1234.jpg, app-profiling_4.gif
>
>
> {noformat}
> = test session starts
> ==
> platform linux -- Python 3.7.3, pytest-5.4.3, py-1.8.2, pluggy-0.13.1
> cachedir: .tox/py37-cython/.pytest_cache
> rootdir: /__w/3/s/flink-python
> collected 568 items
> pyflink/common/tests/test_configuration.py ..[
> 1%]
> pyflink/common/tests/test_execution_config.py ...[
> 5%]
> pyflink/dataset/tests/test_execution_environment.py .
> ##[error]Exit code 137 returned from process: file name '/bin/docker',
> arguments 'exec -i -u 1002
> 97fc4e22522d2ced1f4d23096b8929045d083dd0a99a4233a8b20d0489e9bddb
> /__a/externals/node/bin/node /__w/_temp/containerHandlerInvoker.js'.
> Finishing: Test - python
> {noformat}
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=3729&view=logs&j=9cada3cb-c1d3-5621-16da-0f718fb86602&t=8d78fe4f-d658-5c70-12f8-4921589024c3
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-29461) ProcessDataStreamStreamingTests.test_process_function unstable
[
https://issues.apache.org/jira/browse/FLINK-29461?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640473#comment-17640473
]
Xingbo Huang commented on FLINK-29461:
--
[~mapohl] Sorry for the late reply, I will take a look into this issue in these
two days.
> ProcessDataStreamStreamingTests.test_process_function unstable
> --
>
> Key: FLINK-29461
> URL: https://issues.apache.org/jira/browse/FLINK-29461
> Project: Flink
> Issue Type: Bug
> Components: API / Python
>Affects Versions: 1.16.0, 1.17.0
>Reporter: Huang Xingbo
>Assignee: Huang Xingbo
>Priority: Critical
> Labels: test-stability
>
> {code:java}
> 2022-09-29T02:10:45.3571648Z Sep 29 02:10:45 self =
> testMethod=test_process_function>
> 2022-09-29T02:10:45.3572279Z Sep 29 02:10:45
> 2022-09-29T02:10:45.3572810Z Sep 29 02:10:45 def
> test_process_function(self):
> 2022-09-29T02:10:45.3573495Z Sep 29 02:10:45
> self.env.set_parallelism(1)
> 2022-09-29T02:10:45.3574148Z Sep 29 02:10:45
> self.env.get_config().set_auto_watermark_interval(2000)
> 2022-09-29T02:10:45.3580634Z Sep 29 02:10:45
> self.env.set_stream_time_characteristic(TimeCharacteristic.EventTime)
> 2022-09-29T02:10:45.3583194Z Sep 29 02:10:45 data_stream =
> self.env.from_collection([(1, '1603708211000'),
> 2022-09-29T02:10:45.3584515Z Sep 29 02:10:45
>(2, '1603708224000'),
> 2022-09-29T02:10:45.3585957Z Sep 29 02:10:45
>(3, '1603708226000'),
> 2022-09-29T02:10:45.3587132Z Sep 29 02:10:45
>(4, '1603708289000')],
> 2022-09-29T02:10:45.3588094Z Sep 29 02:10:45
> type_info=Types.ROW([Types.INT(), Types.STRING()]))
> 2022-09-29T02:10:45.3589090Z Sep 29 02:10:45
> 2022-09-29T02:10:45.3589949Z Sep 29 02:10:45 class
> MyProcessFunction(ProcessFunction):
> 2022-09-29T02:10:45.3590710Z Sep 29 02:10:45
> 2022-09-29T02:10:45.3591856Z Sep 29 02:10:45 def
> process_element(self, value, ctx):
> 2022-09-29T02:10:45.3592873Z Sep 29 02:10:45
> current_timestamp = ctx.timestamp()
> 2022-09-29T02:10:45.3593862Z Sep 29 02:10:45
> current_watermark = ctx.timer_service().current_watermark()
> 2022-09-29T02:10:45.3594915Z Sep 29 02:10:45 yield "current
> timestamp: {}, current watermark: {}, current_value: {}"\
> 2022-09-29T02:10:45.3596201Z Sep 29 02:10:45
> .format(str(current_timestamp), str(current_watermark), str(value))
> 2022-09-29T02:10:45.3597089Z Sep 29 02:10:45
> 2022-09-29T02:10:45.3597942Z Sep 29 02:10:45 watermark_strategy =
> WatermarkStrategy.for_monotonous_timestamps()\
> 2022-09-29T02:10:45.3599260Z Sep 29 02:10:45
> .with_timestamp_assigner(SecondColumnTimestampAssigner())
> 2022-09-29T02:10:45.3600611Z Sep 29 02:10:45
> data_stream.assign_timestamps_and_watermarks(watermark_strategy)\
> 2022-09-29T02:10:45.3601877Z Sep 29 02:10:45
> .process(MyProcessFunction(),
> output_type=Types.STRING()).add_sink(self.test_sink)
> 2022-09-29T02:10:45.3603527Z Sep 29 02:10:45 self.env.execute('test
> process function')
> 2022-09-29T02:10:45.3604445Z Sep 29 02:10:45 results =
> self.test_sink.get_results()
> 2022-09-29T02:10:45.3605684Z Sep 29 02:10:45 expected = ["current
> timestamp: 1603708211000, current watermark: "
> 2022-09-29T02:10:45.3607157Z Sep 29 02:10:45
> "-9223372036854775808, current_value: Row(f0=1, f1='1603708211000')",
> 2022-09-29T02:10:45.3608256Z Sep 29 02:10:45 "current
> timestamp: 1603708224000, current watermark: "
> 2022-09-29T02:10:45.3609650Z Sep 29 02:10:45
> "-9223372036854775808, current_value: Row(f0=2, f1='1603708224000')",
> 2022-09-29T02:10:45.3610854Z Sep 29 02:10:45 "current
> timestamp: 1603708226000, current watermark: "
> 2022-09-29T02:10:45.3612279Z Sep 29 02:10:45
> "-9223372036854775808, current_value: Row(f0=3, f1='1603708226000')",
> 2022-09-29T02:10:45.3613382Z Sep 29 02:10:45 "current
> timestamp: 1603708289000, current watermark: "
> 2022-09-29T02:10:45.3615683Z Sep 29 02:10:45
> "-9223372036854775808, current_value: Row(f0=4, f1='1603708289000')"]
> 2022-09-29T02:10:45.3617687Z Sep 29 02:10:45 >
> self.assert_equals_sorted(expected, results)
> 2022-09-29T02:10:45.3618620Z Sep 29 02:10:45
> 2022-09-29T02:10:45.3619425Z Sep 29 02:10:45
> pyflink/datastream/tests/test_data_stream.py:986:
> 2022-09-29T02:10:45.3620424Z Sep 29 02:10:4
[GitHub] [flink-ml] lindong28 commented on a diff in pull request #181: [FLINK-30124] Support collecting model data with arrays and maps
lindong28 commented on code in PR #181:
URL: https://github.com/apache/flink-ml/pull/181#discussion_r1034447969
##
flink-ml-python/pyflink/ml/lib/feature/tests/test_kbinsdiscretizer.py:
##
@@ -163,7 +166,15 @@ def test_get_model_data(self):
expected_field_names = ['binEdges']
self.assertEqual(expected_field_names,
model_data.get_schema().get_field_names())
-# TODO: Add test to collect and verify the model data results after
FLINK-30122 is resolved.
+model_rows = [result for result in
+
self.t_env.to_data_stream(model_data).execute_and_collect()]
+self.assertEqual(1, len(model_rows))
+bin_edges = model_rows[0][expected_field_names.index('binEdges')]
+self.assertEqual(3, len(bin_edges))
+self.assertListEqual([1, 5, 9, 13], bin_edges[0])
+Double = get_gateway().jvm.java.lang.Double
Review Comment:
Can we avoid using get_gateway() here?
Typically python tests should mimic how users can interact with Flink ML
python API. It seems weird that users need to interpret output based on
get_gateway().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] zentol merged pull request #21409: [FLINK-30228][test] Use thread-safe list
zentol merged PR #21409: URL: https://github.com/apache/flink/pull/21409 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-30228) RateLimitedSourceReaderITCase.testRateLimitingParallelExecution failed
[
https://issues.apache.org/jira/browse/FLINK-30228?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chesnay Schepler updated FLINK-30228:
-
Issue Type: Technical Debt (was: Bug)
> RateLimitedSourceReaderITCase.testRateLimitingParallelExecution failed
> --
>
> Key: FLINK-30228
> URL: https://issues.apache.org/jira/browse/FLINK-30228
> Project: Flink
> Issue Type: Technical Debt
> Components: Tests
>Affects Versions: 1.17.0
>Reporter: Matthias Pohl
>Assignee: Chesnay Schepler
>Priority: Major
> Labels: pull-request-available, test-stability
> Fix For: 1.17
>
>
> {{RateLimitedSourceReaderITCase.testRateLimitingParallelExecution}} failed
> due to {{NullPointerException}}:
> {code}
> Nov 28 09:51:10 [ERROR]
> org.apache.flink.api.connector.source.lib.util.RateLimitedSourceReaderITCase.testRateLimitingParallelExecution
> Time elapsed: 1.051 s <<< ERROR!
> Nov 28 09:51:10 java.lang.NullPointerException
> Nov 28 09:51:10 at
> java.util.stream.ReferencePipeline$4$1.accept(ReferencePipeline.java:210)
> Nov 28 09:51:10 at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
> Nov 28 09:51:10 at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
> Nov 28 09:51:10 at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
> Nov 28 09:51:10 at
> java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> Nov 28 09:51:10 at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> Nov 28 09:51:10 at
> java.util.stream.IntPipeline.reduce(IntPipeline.java:479)
> Nov 28 09:51:10 at
> java.util.stream.IntPipeline.sum(IntPipeline.java:437)
> Nov 28 09:51:10 at
> org.apache.flink.api.connector.source.lib.util.RateLimitedSourceReaderITCase$MockRateLimiterStrategy.getRateLimitersCallCount(RateLimitedSourceReaderITCase.java:116)
> Nov 28 09:51:10 at
> org.apache.flink.api.connector.source.lib.util.RateLimitedSourceReaderITCase.testRateLimitingParallelExecution(RateLimitedSourceReaderITCase.java:77)
> {code}
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=43541&view=logs&j=5c8e7682-d68f-54d1-16a2-a09310218a49&t=86f654fa-ab48-5c1a-25f4-7e7f6afb9bba&l=7016
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Closed] (FLINK-30228) RateLimitedSourceReaderITCase.testRateLimitingParallelExecution failed
[
https://issues.apache.org/jira/browse/FLINK-30228?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chesnay Schepler closed FLINK-30228.
Fix Version/s: 1.17
Resolution: Fixed
master: 3130b9a907afb420f95a64a3cada19156fedc72a
> RateLimitedSourceReaderITCase.testRateLimitingParallelExecution failed
> --
>
> Key: FLINK-30228
> URL: https://issues.apache.org/jira/browse/FLINK-30228
> Project: Flink
> Issue Type: Bug
> Components: Tests
>Affects Versions: 1.17.0
>Reporter: Matthias Pohl
>Assignee: Chesnay Schepler
>Priority: Major
> Labels: pull-request-available, test-stability
> Fix For: 1.17
>
>
> {{RateLimitedSourceReaderITCase.testRateLimitingParallelExecution}} failed
> due to {{NullPointerException}}:
> {code}
> Nov 28 09:51:10 [ERROR]
> org.apache.flink.api.connector.source.lib.util.RateLimitedSourceReaderITCase.testRateLimitingParallelExecution
> Time elapsed: 1.051 s <<< ERROR!
> Nov 28 09:51:10 java.lang.NullPointerException
> Nov 28 09:51:10 at
> java.util.stream.ReferencePipeline$4$1.accept(ReferencePipeline.java:210)
> Nov 28 09:51:10 at
> java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
> Nov 28 09:51:10 at
> java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
> Nov 28 09:51:10 at
> java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
> Nov 28 09:51:10 at
> java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
> Nov 28 09:51:10 at
> java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
> Nov 28 09:51:10 at
> java.util.stream.IntPipeline.reduce(IntPipeline.java:479)
> Nov 28 09:51:10 at
> java.util.stream.IntPipeline.sum(IntPipeline.java:437)
> Nov 28 09:51:10 at
> org.apache.flink.api.connector.source.lib.util.RateLimitedSourceReaderITCase$MockRateLimiterStrategy.getRateLimitersCallCount(RateLimitedSourceReaderITCase.java:116)
> Nov 28 09:51:10 at
> org.apache.flink.api.connector.source.lib.util.RateLimitedSourceReaderITCase.testRateLimitingParallelExecution(RateLimitedSourceReaderITCase.java:77)
> {code}
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=43541&view=logs&j=5c8e7682-d68f-54d1-16a2-a09310218a49&t=86f654fa-ab48-5c1a-25f4-7e7f6afb9bba&l=7016
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[GitHub] [flink] zentol opened a new pull request, #21416: [FLINK-30202][tests] Do not assert on checkpointId
zentol opened a new pull request, #21416: URL: https://github.com/apache/flink/pull/21416 Capturing the checkpointId for a generated record in a subsequent map function is impossible since the notifyCheckpointComplete notification may arrive at any time (or not at all). Instead just assert that each subtask got exactly as many records as expected, which can only happen (reliably) if the rate-limiting works as expected. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-30202) DataGeneratorSourceITCase.testGatedRateLimiter failed
[
https://issues.apache.org/jira/browse/FLINK-30202?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-30202:
---
Labels: pull-request-available test-stability (was: test-stability)
> DataGeneratorSourceITCase.testGatedRateLimiter failed
> -
>
> Key: FLINK-30202
> URL: https://issues.apache.org/jira/browse/FLINK-30202
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Common
>Affects Versions: 1.17.0
>Reporter: Matthias Pohl
>Assignee: Chesnay Schepler
>Priority: Major
> Labels: pull-request-available, test-stability
>
> [This
> build|https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=43483&view=logs&j=5cae8624-c7eb-5c51-92d3-4d2dacedd221&t=5acec1b4-945b-59ca-34f8-168928ce5199&l=24747]
> failed due to a failed assertion in
> {{{}DataGeneratorSourceITCase.testGatedRateLimiter{}}}:
> {code:java}
> Nov 25 03:26:45 org.opentest4j.AssertionFailedError:
> Nov 25 03:26:45
> Nov 25 03:26:45 expected: 2
> Nov 25 03:26:45 but was: 1 {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Updated] (FLINK-30202) DataGeneratorSourceITCase.testGatedRateLimiter failed
[
https://issues.apache.org/jira/browse/FLINK-30202?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Chesnay Schepler updated FLINK-30202:
-
Issue Type: Technical Debt (was: Bug)
> DataGeneratorSourceITCase.testGatedRateLimiter failed
> -
>
> Key: FLINK-30202
> URL: https://issues.apache.org/jira/browse/FLINK-30202
> Project: Flink
> Issue Type: Technical Debt
> Components: Connectors / Common
>Affects Versions: 1.17.0
>Reporter: Matthias Pohl
>Assignee: Chesnay Schepler
>Priority: Major
> Labels: pull-request-available, test-stability
>
> [This
> build|https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=43483&view=logs&j=5cae8624-c7eb-5c51-92d3-4d2dacedd221&t=5acec1b4-945b-59ca-34f8-168928ce5199&l=24747]
> failed due to a failed assertion in
> {{{}DataGeneratorSourceITCase.testGatedRateLimiter{}}}:
> {code:java}
> Nov 25 03:26:45 org.opentest4j.AssertionFailedError:
> Nov 25 03:26:45
> Nov 25 03:26:45 expected: 2
> Nov 25 03:26:45 but was: 1 {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[GitHub] [flink-table-store] JingsongLi commented on pull request #344: [FLINK-29823] Support to get schema of table snapshot
JingsongLi commented on PR #344: URL: https://github.com/apache/flink-table-store/pull/344#issuecomment-1330296296 > Hi @JingsongLi Sorry for replying late. I think `MyTable$schemas` is so good that we cannot to check the schema change history from Table Store currently. I think we can create a new issue for it. > > Back to this PR, what I want is that users can query data from Table Store with given snapshot id. Do you think it is better for users to add parameters in options, or to specify the snapshot ID on the table name in query? THX I think users can just `select a.* from MyTable$schemas a, MyTable$snapshots b where snapshot_id=10 and a.schema_id = b.schema_id` to get the schema for one snapshot id. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21416: [FLINK-30202][tests] Do not assert on checkpointId
flinkbot commented on PR #21416: URL: https://github.com/apache/flink/pull/21416#issuecomment-1330298212 ## CI report: * b372c77de18aef2ceea6b00ad64a3278fd91a8d1 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] zentol opened a new pull request, #21417: [FLINK-29862] Upgrade to flink-shaded 16.1
zentol opened a new pull request, #21417: URL: https://github.com/apache/flink/pull/21417 Second attempt. Identical changes to #21227. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #21417: [FLINK-29862] Upgrade to flink-shaded 16.1
flinkbot commented on PR #21417: URL: https://github.com/apache/flink/pull/21417#issuecomment-1330311573 ## CI report: * 565e4934f4ca7357c75a6a10ead225fe900ba996 UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Created] (FLINK-30237) Only bundle a single Zookeeper version
Chesnay Schepler created FLINK-30237: Summary: Only bundle a single Zookeeper version Key: FLINK-30237 URL: https://issues.apache.org/jira/browse/FLINK-30237 Project: Flink Issue Type: Technical Debt Components: Runtime / Coordination Reporter: Chesnay Schepler Assignee: Chesnay Schepler Fix For: 1.17.0 Way back when we added support for ZK 3.5 we started bundling 2 zookeeper clients, because of incompatibilities between 3.4 and 3.5. This is no longer required, and we could simplify things again. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink-ml] lindong28 commented on pull request #181: [FLINK-30124] Support collecting model data with arrays and maps
lindong28 commented on PR #181: URL: https://github.com/apache/flink-ml/pull/181#issuecomment-1330320243 Thanks for the update. LGTM. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-ml] lindong28 merged pull request #181: [FLINK-30124] Support collecting model data with arrays and maps
lindong28 merged PR #181: URL: https://github.com/apache/flink-ml/pull/181 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] zentol opened a new pull request, #21418: [FLINK-30237][coordination] Only bundle single ZK client
zentol opened a new pull request, #21418: URL: https://github.com/apache/flink/pull/21418 Reverts FLINK-13417 and FLINK-16350 since it's no longer required to bundle 2 different clients. We'll upgrade to 3.7 in FLINK-29420. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-29420) Upgrade Zookeeper to 3.7
[ https://issues.apache.org/jira/browse/FLINK-29420?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Chesnay Schepler updated FLINK-29420: - Summary: Upgrade Zookeeper to 3.7 (was: Upgrade Zookeeper to 3.7 / 3.8) > Upgrade Zookeeper to 3.7 > > > Key: FLINK-29420 > URL: https://issues.apache.org/jira/browse/FLINK-29420 > Project: Flink > Issue Type: Improvement > Components: BuildSystem / Shaded, Runtime / Coordination >Reporter: Chesnay Schepler >Assignee: Chesnay Schepler >Priority: Major > Fix For: 1.17.0 > > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-30237) Only bundle a single Zookeeper version
[ https://issues.apache.org/jira/browse/FLINK-30237?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-30237: --- Labels: pull-request-available (was: ) > Only bundle a single Zookeeper version > -- > > Key: FLINK-30237 > URL: https://issues.apache.org/jira/browse/FLINK-30237 > Project: Flink > Issue Type: Technical Debt > Components: Runtime / Coordination >Reporter: Chesnay Schepler >Assignee: Chesnay Schepler >Priority: Major > Labels: pull-request-available > Fix For: 1.17.0 > > > Way back when we added support for ZK 3.5 we started bundling 2 zookeeper > clients, because of incompatibilities between 3.4 and 3.5. This is no longer > required, and we could simplify things again. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Updated] (FLINK-30124) GenericType is not supported in PyFlink currently
[
https://issues.apache.org/jira/browse/FLINK-30124?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dong Lin updated FLINK-30124:
-
Fix Version/s: ml-2.2.0
> GenericType is not supported in PyFlink currently
>
>
> Key: FLINK-30124
> URL: https://issues.apache.org/jira/browse/FLINK-30124
> Project: Flink
> Issue Type: Bug
> Components: Library / Machine Learning
>Affects Versions: ml-2.1.0
>Reporter: Yunfeng Zhou
>Assignee: Yunfeng Zhou
>Priority: Major
> Labels: pull-request-available
> Fix For: ml-2.2.0
>
>
> When we add and execute the following test case to
> flink-ml-python/pyflink/ml/lib/classification/tests/test_naivebayes.py of the
> Flink ML repository,
> {code:java}
> def test_get_model_data(self):
> model_data = self.estimator.fit(self.train_data).get_model_data()[0]
> self.t_env.to_data_stream(model_data).execute_and_collect().next(){code}
> The following exception would be thrown.
>
> {code:java}
> j_type_info = JavaObject id=o698
> def _from_java_type(j_type_info: JavaObject) -> TypeInformation:
> gateway = get_gateway()
> JBasicTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo
>
> if _is_instance_of(j_type_info, JBasicTypeInfo.STRING_TYPE_INFO):
> return Types.STRING()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BOOLEAN_TYPE_INFO):
> return Types.BOOLEAN()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BYTE_TYPE_INFO):
> return Types.BYTE()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.SHORT_TYPE_INFO):
> return Types.SHORT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.INT_TYPE_INFO):
> return Types.INT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.LONG_TYPE_INFO):
> return Types.LONG()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.FLOAT_TYPE_INFO):
> return Types.FLOAT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.DOUBLE_TYPE_INFO):
> return Types.DOUBLE()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.CHAR_TYPE_INFO):
> return Types.CHAR()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_INT_TYPE_INFO):
> return Types.BIG_INT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_DEC_TYPE_INFO):
> return Types.BIG_DEC()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.INSTANT_TYPE_INFO):
> return Types.INSTANT()
>
> JSqlTimeTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo
> if _is_instance_of(j_type_info, JSqlTimeTypeInfo.DATE):
> return Types.SQL_DATE()
> elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIME):
> return Types.SQL_TIME()
> elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIMESTAMP):
> return Types.SQL_TIMESTAMP()
>
> JPrimitiveArrayTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo \
> .PrimitiveArrayTypeInfo
>
> if _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.BOOLEAN())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.BYTE())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.SHORT())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.INT())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.LONG())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.FLOAT())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.DOUBLE())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.CHAR())
>
> JBasicArrayTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo
>
> if _is_instance_of(j_type_info,
> JBasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.BOOLEAN())
> elif _is_instance_of(j_type_info,
> JBasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO):
>
[jira] [Resolved] (FLINK-30124) GenericType is not supported in PyFlink currently
[
https://issues.apache.org/jira/browse/FLINK-30124?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Dong Lin resolved FLINK-30124.
--
Resolution: Fixed
> GenericType is not supported in PyFlink currently
>
>
> Key: FLINK-30124
> URL: https://issues.apache.org/jira/browse/FLINK-30124
> Project: Flink
> Issue Type: Bug
> Components: Library / Machine Learning
>Affects Versions: ml-2.1.0
>Reporter: Yunfeng Zhou
>Assignee: Yunfeng Zhou
>Priority: Major
> Labels: pull-request-available
>
> When we add and execute the following test case to
> flink-ml-python/pyflink/ml/lib/classification/tests/test_naivebayes.py of the
> Flink ML repository,
> {code:java}
> def test_get_model_data(self):
> model_data = self.estimator.fit(self.train_data).get_model_data()[0]
> self.t_env.to_data_stream(model_data).execute_and_collect().next(){code}
> The following exception would be thrown.
>
> {code:java}
> j_type_info = JavaObject id=o698
> def _from_java_type(j_type_info: JavaObject) -> TypeInformation:
> gateway = get_gateway()
> JBasicTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo.BasicTypeInfo
>
> if _is_instance_of(j_type_info, JBasicTypeInfo.STRING_TYPE_INFO):
> return Types.STRING()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BOOLEAN_TYPE_INFO):
> return Types.BOOLEAN()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BYTE_TYPE_INFO):
> return Types.BYTE()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.SHORT_TYPE_INFO):
> return Types.SHORT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.INT_TYPE_INFO):
> return Types.INT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.LONG_TYPE_INFO):
> return Types.LONG()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.FLOAT_TYPE_INFO):
> return Types.FLOAT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.DOUBLE_TYPE_INFO):
> return Types.DOUBLE()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.CHAR_TYPE_INFO):
> return Types.CHAR()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_INT_TYPE_INFO):
> return Types.BIG_INT()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.BIG_DEC_TYPE_INFO):
> return Types.BIG_DEC()
> elif _is_instance_of(j_type_info, JBasicTypeInfo.INSTANT_TYPE_INFO):
> return Types.INSTANT()
>
> JSqlTimeTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo.SqlTimeTypeInfo
> if _is_instance_of(j_type_info, JSqlTimeTypeInfo.DATE):
> return Types.SQL_DATE()
> elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIME):
> return Types.SQL_TIME()
> elif _is_instance_of(j_type_info, JSqlTimeTypeInfo.TIMESTAMP):
> return Types.SQL_TIMESTAMP()
>
> JPrimitiveArrayTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo \
> .PrimitiveArrayTypeInfo
>
> if _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.BOOLEAN_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.BOOLEAN())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.BYTE_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.BYTE())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.SHORT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.SHORT())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.INT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.INT())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.LONG_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.LONG())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.FLOAT_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.FLOAT())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.DOUBLE_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.DOUBLE())
> elif _is_instance_of(j_type_info,
> JPrimitiveArrayTypeInfo.CHAR_PRIMITIVE_ARRAY_TYPE_INFO):
> return Types.PRIMITIVE_ARRAY(Types.CHAR())
>
> JBasicArrayTypeInfo =
> gateway.jvm.org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo
>
> if _is_instance_of(j_type_info,
> JBasicArrayTypeInfo.BOOLEAN_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(Types.BOOLEAN())
> elif _is_instance_of(j_type_info,
> JBasicArrayTypeInfo.BYTE_ARRAY_TYPE_INFO):
> return Types.BASIC_ARRAY(T
[GitHub] [flink] flinkbot commented on pull request #21418: [FLINK-30237][coordination] Only bundle single ZK client
flinkbot commented on PR #21418: URL: https://github.com/apache/flink/pull/21418#issuecomment-1330324897 ## CI report: * 914511a382988333dfd9d0dfcc28158d197660ed UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-connector-aws] dannycranmer opened a new pull request, #32: Flink 29907[FLINK-29907][Connectors/AWS] Externalize KDS/KDF and AWS base from Flink repo
dannycranmer opened a new pull request, #32: URL: https://github.com/apache/flink-connector-aws/pull/32 TODO -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Updated] (FLINK-29907) Externalize AWS connectors from Flink core
[
https://issues.apache.org/jira/browse/FLINK-29907?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-29907:
---
Labels: pull-request-available (was: )
> Externalize AWS connectors from Flink core
> --
>
> Key: FLINK-29907
> URL: https://issues.apache.org/jira/browse/FLINK-29907
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / AWS
>Reporter: Danny Cranmer
>Assignee: Danny Cranmer
>Priority: Major
> Labels: pull-request-available
> Fix For: aws-connector-4.0.0
>
>
> Externlize the following modules from Flink core to the connectors repo:
> - {{flink-connector-aws-base}}
> - {{flink-connector-kinesis}}
> - {{flink-connector-sql-kinesis}}
> - {{flink-connector-aws-kinesis-streams}}
> - {{flink-connector-sql-aws-kinesis-streams}}
> - {{flink-connector-aws-kinesis-firehose}}
> - {{flink-connector-sql-aws-kinesis-firehose}}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-30052) Move existing Kafka connector code from Flink repo to dedicated Kafka repo
[ https://issues.apache.org/jira/browse/FLINK-30052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640506#comment-17640506 ] Mason Chen commented on FLINK-30052: [~martijnvisser] I started this task over the weekend. There were some issues in porting the Kafka related E2E tests over, using 1.17-SNAPSHOT for the Flink version, since the snapshot artifacts do not exist for flink-sql-client-test [1] and flink-end-to-end-tests-common [2]. [1] is built at build time using the `maven-dependency-plugin`. I suspect that [1] is something that the other connectors will face and can be solved commonly. With regards to flink-to-end-tests-common, should a snapshot be published here: [https://repository.apache.org/content/groups/snapshots/org/apache/flink/]? Elasticsearch E2E tests fortunately didn't have these dependencies but other connectors do. Also, due to the changes in 1.17 (split level watermark alignment), it isn't straightforward to downgrade the master code to 1.16, so I will need to repeat the externalization instructions for master branch, release-1.16 branch, etc. [1] [https://github.com/apache/flink/blob/bee3e9442fa2749f0b2e21d79949b5410fa422e5/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml#L196] [2] [https://github.com/apache/flink/blob/bee3e9442fa2749f0b2e21d79949b5410fa422e5/flink-end-to-end-tests/flink-end-to-end-tests-common-kafka/pom.xml#L43] > Move existing Kafka connector code from Flink repo to dedicated Kafka repo > -- > > Key: FLINK-30052 > URL: https://issues.apache.org/jira/browse/FLINK-30052 > Project: Flink > Issue Type: Sub-task > Components: Connectors / Kafka >Reporter: Martijn Visser >Assignee: Mason Chen >Priority: Major > > Instructions guide can be found at > https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] zentol commented on a diff in pull request #21128: [FLINK-29710] Bump minimum supported Hadoop version to 2.10.2
zentol commented on code in PR #21128:
URL: https://github.com/apache/flink/pull/21128#discussion_r1034500893
##
flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java:
##
@@ -608,10 +611,13 @@ public boolean accept(File dir, String name) {
public static boolean verifyStringsInNamedLogFiles(
final String[] mustHave, final ApplicationId applicationId, final
String fileName) {
final List mustHaveList = Arrays.asList(mustHave);
-final File cwd = new File("target",
YARN_CONFIGURATION.get(TEST_CLUSTER_NAME_KEY));
+final File cwd =
Review Comment:
Why was this failing on 2.10, when it worked on 2.8 and 3.3?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[jira] [Commented] (FLINK-30220) Secure user credentials in Flink SQL JDBC connector
[
https://issues.apache.org/jira/browse/FLINK-30220?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640511#comment-17640511
]
Jun Qin commented on FLINK-30220:
-
The same is needed for JDBC connector:
CREATE TABLE MyUserTable (
id BIGINT,
name STRING,
age INT,
status BOOLEAN,
PRIMARY KEY (id) NOT ENFORCED
) WITH ( 'connector' = 'jdbc', 'url' =
'jdbc:mysql://localhost:3306/mydatabase', 'table-name' = 'users',
'username' = 'a-username', 'password' = 'a-password');
> Secure user credentials in Flink SQL JDBC connector
> ---
>
> Key: FLINK-30220
> URL: https://issues.apache.org/jira/browse/FLINK-30220
> Project: Flink
> Issue Type: Improvement
>Reporter: Jun Qin
>Priority: Major
>
> Similar to FLINK-28028, when using Flink SQL JDBC connector, we should also
> have a way to secure the username and the password used in the DDL:
> {code:java}
> CREATE TABLE MyUserTable (
> id BIGINT,
> name STRING,
> age INT,
> status BOOLEAN,
> PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
>'connector' = 'jdbc',
>'url' = 'jdbc:mysql://localhost:3306/mydatabase',
>'table-name' = 'users',
>'username' = 'a-username',
>'password' = 'a-password'
> );
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-30052) Move existing Kafka connector code from Flink repo to dedicated Kafka repo
[ https://issues.apache.org/jira/browse/FLINK-30052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640571#comment-17640571 ] Martijn Visser commented on FLINK-30052: [~mason6345] The best course of actions is to first move the code from "release-1.16" to the external repository and create a release "3.0.0" from that. When that's done, then to move over the code from "master" so that can be released as "3.1.0" when Flink 1.17 is released. > Move existing Kafka connector code from Flink repo to dedicated Kafka repo > -- > > Key: FLINK-30052 > URL: https://issues.apache.org/jira/browse/FLINK-30052 > Project: Flink > Issue Type: Sub-task > Components: Connectors / Kafka >Reporter: Martijn Visser >Assignee: Mason Chen >Priority: Major > > Instructions guide can be found at > https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-30052) Move existing Kafka connector code from Flink repo to dedicated Kafka repo
[ https://issues.apache.org/jira/browse/FLINK-30052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640573#comment-17640573 ] Martijn Visser commented on FLINK-30052: [~chesnay] WDYT on the issues that Mason mentioned? > Move existing Kafka connector code from Flink repo to dedicated Kafka repo > -- > > Key: FLINK-30052 > URL: https://issues.apache.org/jira/browse/FLINK-30052 > Project: Flink > Issue Type: Sub-task > Components: Connectors / Kafka >Reporter: Martijn Visser >Assignee: Mason Chen >Priority: Major > > Instructions guide can be found at > https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] MartijnVisser commented on a diff in pull request #21128: [FLINK-29710] Bump minimum supported Hadoop version to 2.10.2
MartijnVisser commented on code in PR #21128:
URL: https://github.com/apache/flink/pull/21128#discussion_r1034520479
##
flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java:
##
@@ -608,10 +611,13 @@ public boolean accept(File dir, String name) {
public static boolean verifyStringsInNamedLogFiles(
final String[] mustHave, final ApplicationId applicationId, final
String fileName) {
final List mustHaveList = Arrays.asList(mustHave);
-final File cwd = new File("target",
YARN_CONFIGURATION.get(TEST_CLUSTER_NAME_KEY));
+final File cwd =
Review Comment:
@gaborgsomogyi ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-connector-jdbc] zentol commented on a diff in pull request #1: [FLINK-30060][Connector/JDBC] Move existing JDBC connector code from Flink repo to dedicated JDBC repo
zentol commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-jdbc/pull/1#discussion_r1034507671
##
.idea/vcs.xml:
##
@@ -0,0 +1,6 @@
+
+
+
Review Comment:
There's a better vcs.xml example in the ES repo.
##
pom.xml:
##
@@ -0,0 +1,421 @@
+
+
+http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";
+xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";>
+
+
+io.github.zentol.flink
+flink-connector-parent
+1.0
+
+
+4.0.0
+
+org.apache.flink
+flink-connector-jdbc-parent
+3.0-SNAPSHOT
+Flink : Connectors : JDBC : Parent
+pom
+2022
+
+
+https://github.com/apache/flink-connector-jdbc
+
g...@github.com:apache/flink-connector-jdbc.git
+
+
scm:git:https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
+
+
+
+
+flink-connector-jdbc
+
+
+
+1.16.0
+15.0
+
+2.13.4.20221013
+4.13.2
+5.8.1
+3.21.0
+0.22.0
+1.17.2
+2.21.0
+
+false
+1.15.0
+
+1.7.36
+2.17.2
+
+
Review Comment:
Is this actually used by any jdbc test?
##
pom.xml:
##
@@ -0,0 +1,421 @@
+
+
+http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";
+xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";>
+
+
+io.github.zentol.flink
+flink-connector-parent
+1.0
+
+
+4.0.0
+
+org.apache.flink
+flink-connector-jdbc-parent
+3.0-SNAPSHOT
+Flink : Connectors : JDBC : Parent
+pom
+2022
+
+
+https://github.com/apache/flink-connector-jdbc
+
g...@github.com:apache/flink-connector-jdbc.git
+
+
scm:git:https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
+
+
+
+
+flink-connector-jdbc
+
+
+
+1.16.0
+15.0
+
+2.13.4.20221013
+4.13.2
+5.8.1
+3.21.0
+0.22.0
+1.17.2
+2.21.0
+
+false
+1.15.0
+
+1.7.36
+2.17.2
+
+
+
+
flink-connector-jdbc-parent
+
+
+
+
+org.apache.flink
+flink-shaded-force-shading
+${flink.shaded.version}
+
+
+
+
+
+
+org.slf4j
+slf4j-api
+provided
+
+
+
+
+com.google.code.findbugs
+jsr305
+provided
+
+
+
+
+org.junit.jupiter
+junit-jupiter
+test
+
+
+
+org.junit.vintage
+junit-vintage-engine
+test
+
+
+
+org.assertj
+assertj-core
+test
+
+
+
+org.mockito
+mockito-core
+${mockito.version}
+jar
+test
+
+
+
+org.testcontainers
+junit-jupiter
+test
+
+
+
+
+
+org.apache.logging.log4j
+log4j-slf4j-impl
+test
+
+
+
+org.apache.logging.log4j
+log4j-api
+test
+
+
+
+org.apache.logging.log4j
+log4j-core
+test
+
+
+
+
+org.apache.logging.log4j
+log4j-1.2-api
+test
+
+
+
+org.apache.flink
+flink-test-utils-junit
+test
+
+
+
+
+org.apache.flink
+flink-architecture-tests-test
+test
+
+
+org.apache.flink
+flink-architecture-tests-production
+test
+
+
+
+
+
+
+
+
+
+
+
+
+org.apache.flink
+flink-test-utils
+${flink.version}
+test
+
+
+log4j
+log4j
+
+
+org.slf4j
+slf4j-log4j12
+
+
Review Comment:
shouldnt be required since 1.16.0
##
pom.xml:
##
@@ -0,0 +1,421 @@
+
+
+http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";
+xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";>
+
+
+io.github.zentol.flink
+flink-connector-parent
+1.0
+
+
+4.0.0
+
+org.apache.flink
+flink-connector-jdbc-parent
+3.0-SNAPSHOT
+Flink : Connectors
[GitHub] [flink-connector-pulsar] zentol commented on a diff in pull request #1: [FLINK-30054][Connector/Pulsar] Move existing Pulsar connector code
zentol commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-pulsar/pull/1#discussion_r1034527661
##
pom.xml:
##
@@ -0,0 +1,417 @@
+
+
+http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";
+xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";>
+
+
+io.github.zentol.flink
+flink-connector-parent
+1.0
+
+
+4.0.0
+
+org.apache.flink
+flink-connector-pulsar-parent
+3.0-SNAPSHOT
+Flink : Connectors : Pulsar : Parent
+pom
+2022
+
+
+https://github.com/apache/flink-connector-pulsar
+
g...@github.com:apache/flink-connector-pulsar.git
+
+
scm:git:https://gitbox.apache.org/repos/asf/flink-connector-pulsar.git
+
+
+
+
+flink-connector-pulsar
+flink-sql-connector-pulsar
+
+
+
+1.16.0
+15.0
+
+4.13.2
+5.8.1
+3.23.1
+0.22.0
+1.17.2
+2.21.0
+
+false
+1.15.0
+
+1.7.36
+2.17.2
+
+
+
+
flink-connector-pulsar-parent
+
+
+
+
+org.apache.flink
+flink-shaded-force-shading
+${flink.shaded.version}
+
+
+
+
+
+
+org.slf4j
+slf4j-api
+provided
+
+
+
+
+com.google.code.findbugs
+jsr305
+provided
+
+
+
+
+org.junit.jupiter
+junit-jupiter
+${junit5.version}
+test
+
+
+
+org.junit.vintage
+junit-vintage-engine
+${junit5.version}
+test
+
+
+
+org.assertj
+assertj-core
+${assertj.version}
+test
+
+
+
+org.mockito
+mockito-core
+${mockito.version}
+jar
+test
+
+
+
+org.testcontainers
+junit-jupiter
+test
+
+
+
+
+
+org.apache.logging.log4j
+log4j-slf4j-impl
+test
+
+
+
+org.apache.logging.log4j
+log4j-api
+test
+
+
+
+org.apache.logging.log4j
+log4j-core
+test
+
+
+
+
+org.apache.logging.log4j
+log4j-1.2-api
+test
+
+
+
+org.apache.flink
+flink-test-utils-junit
+test
+
+
+
+
+org.apache.flink
+flink-architecture-tests-test
+test
+
+
+org.apache.flink
+flink-architecture-tests-production
+test
+
+
+
+
+
+
+
+
+
+
+
+
+org.apache.flink
+flink-test-utils
+${flink.version}
+test
+
+
+log4j
+log4j
+
+
+org.slf4j
+slf4j-log4j12
+
+
+
+
+
+org.apache.flink
+flink-test-utils-junit
+${flink.version}
+test
+
+
+
+
+
+org.apache.flink
+flink-architecture-tests-base
+${flink.version}
+test
+
+
+
+org.apache.flink
+flink-architecture-tests-test
+${flink.version}
+test
+
+
+
+org.apache.flink
+flink-architecture-tests-production
+${flink.version}
+test
+
+
+
+
+com.google.code.findbugs
+jsr305
+1.3.9
+
+
+
+commons-codec
+commons-codec
+1.15
+
+
+
+org.apache.httpcomponents
+httpcore
+4.4.14
+
+
+
+org.apache.httpcomponents
+httpclient
+4.5.13
+
+
+
+org.slf4j
+slf4j-api
+${slf4j.version}
+
+
+
+org.apache.logging.log4j
+log4j-slf4j-impl
+${log4j.version}
+
+
+
+org.apache.logging.log4j
+log4j-api
+${log4j.version}
+
[GitHub] [flink-table-store] zjureel commented on pull request #347: [FLINK-27847] Support rename/drop column in SchemaManager
zjureel commented on PR #347: URL: https://github.com/apache/flink-table-store/pull/347#issuecomment-1330416747 Hi @JingsongLi @tsreaper I have update the `SchemaManager` and add more test cases in `SchemaEvolutionTest` and `SparkReadITCase` for schema evolution, please help to review them when you're free, THX :) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Commented] (FLINK-30052) Move existing Kafka connector code from Flink repo to dedicated Kafka repo
[
https://issues.apache.org/jira/browse/FLINK-30052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640592#comment-17640592
]
Chesnay Schepler commented on FLINK-30052:
--
Just replace the {{flink-end-to-end-test-common}} dependency with
{{flink-connector-test-utils}}.
> Move existing Kafka connector code from Flink repo to dedicated Kafka repo
> --
>
> Key: FLINK-30052
> URL: https://issues.apache.org/jira/browse/FLINK-30052
> Project: Flink
> Issue Type: Sub-task
> Components: Connectors / Kafka
>Reporter: Martijn Visser
>Assignee: Mason Chen
>Priority: Major
>
> Instructions guide can be found at
> https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Comment Edited] (FLINK-30052) Move existing Kafka connector code from Flink repo to dedicated Kafka repo
[
https://issues.apache.org/jira/browse/FLINK-30052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640592#comment-17640592
]
Chesnay Schepler edited comment on FLINK-30052 at 11/29/22 10:34 AM:
-
Just replace the {{flink-end-to-end-test-common}} dependency with
{{flink-connector-test-utils}} for flink-sql-client-test.
was (Author: zentol):
Just replace the {{flink-end-to-end-test-common}} dependency with
{{flink-connector-test-utils}}.
> Move existing Kafka connector code from Flink repo to dedicated Kafka repo
> --
>
> Key: FLINK-30052
> URL: https://issues.apache.org/jira/browse/FLINK-30052
> Project: Flink
> Issue Type: Sub-task
> Components: Connectors / Kafka
>Reporter: Martijn Visser
>Assignee: Mason Chen
>Priority: Major
>
> Instructions guide can be found at
> https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[jira] [Commented] (FLINK-30052) Move existing Kafka connector code from Flink repo to dedicated Kafka repo
[ https://issues.apache.org/jira/browse/FLINK-30052?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640593#comment-17640593 ] Chesnay Schepler commented on FLINK-30052: -- Other tests that rely on flink-end-to-end-tests-common should be re-written. > Move existing Kafka connector code from Flink repo to dedicated Kafka repo > -- > > Key: FLINK-30052 > URL: https://issues.apache.org/jira/browse/FLINK-30052 > Project: Flink > Issue Type: Sub-task > Components: Connectors / Kafka >Reporter: Martijn Visser >Assignee: Mason Chen >Priority: Major > > Instructions guide can be found at > https://cwiki.apache.org/confluence/display/FLINK/Externalized+Connector+development -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] huwh commented on pull request #21233: [FLINK-29870] [ResourceManager]Split ResourceActions to ResourceAllocator and ResourceEventListener.
huwh commented on PR #21233: URL: https://github.com/apache/flink/pull/21233#issuecomment-1330443060 > Well, I meant removing `notifyAllocationFailure` in a separate hotfix commit in this PR, to keep the commit history clean. Never mind. I Have merged the PRfor [FLINK-30084](https://issues.apache.org/jira/browse/FLINK-30084). > > Could you please rebase this PR and re-organize the commits. When addressing comments, it would be nice to keep the commits organized as to be merged. You can either address comments in fixup commits that clearly shows which previous commits they should be squashed into, or modify the previous commits and make a force-push if the comments cannot easily be addressed with fixup commits. Hi, @xintongsong, I've rebased this PR, but it's not easy to automatically squash with fixup commits because there are lots of classes rename. So, I modified the previous commit to addressing comments. Could you please review this again in your free time. Thanks a lot. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-table-store] zjureel commented on pull request #402: [FLINK-29988] Using lower case fields in hive catalog
zjureel commented on PR #402: URL: https://github.com/apache/flink-table-store/pull/402#issuecomment-1330449444 After discussed with @JingsongLi , I'll create another issue to throw exception when user use `HiveCatalog` and create table with upper case table and column name. We will fix this issue when we find a better way for `HiveCatalog`. So I close this PR first -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-table-store] zjureel closed pull request #402: [FLINK-29988] Using lower case fields in hive catalog
zjureel closed pull request #402: [FLINK-29988] Using lower case fields in hive catalog URL: https://github.com/apache/flink-table-store/pull/402 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Created] (FLINK-30238) Unified Sink committer does not clean up state on final savepoint
Fabian Paul created FLINK-30238: --- Summary: Unified Sink committer does not clean up state on final savepoint Key: FLINK-30238 URL: https://issues.apache.org/jira/browse/FLINK-30238 Project: Flink Issue Type: Bug Components: Connectors / Common Affects Versions: 1.15.3, 1.17.0, 1.16.1 Reporter: Fabian Paul During stop-with-savepoint the committer only commits the pending committables on notifyCheckpointComplete. This has several downsides. * Last committableSummary has checkpoint id LONG.MAX and is never cleared from the state leading to that stop-with-savepoint does not work when the pipeline recovers from a savepoint * While the committables are committed during stop-with-savepoint they are not forwarded to post-commit topology, potentially losing data and preventing to close open transactions. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink-connector-jdbc] zentol commented on a diff in pull request #1: [FLINK-30060][Connector/JDBC] Move existing JDBC connector code from Flink repo to dedicated JDBC repo
zentol commented on code in PR #1: URL: https://github.com/apache/flink-connector-jdbc/pull/1#discussion_r1034606733 ## pom.xml: ## @@ -0,0 +1,395 @@ + + +http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"; +xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";> + + +io.github.zentol.flink +flink-connector-parent +1.0 + + +4.0.0 + +org.apache.flink +flink-connector-jdbc-parent +3.0-SNAPSHOT +Flink : Connectors : JDBC : Parent +pom +2022 + + +https://github.com/apache/flink-connector-jdbc + g...@github.com:apache/flink-connector-jdbc.git + + scm:git:https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git + + + + +flink-connector-jdbc + + + +1.16.0 +15.0 + +2.13.4.20221013 +4.13.2 +5.8.1 +3.21.0 +0.22.0 +1.17.2 +2.21.0 + +false Review Comment: We shouldn't need this here. This is only really useful in sql-jar modules. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Comment Edited] (FLINK-15656) Support user-specified pod templates
[ https://issues.apache.org/jira/browse/FLINK-15656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640626#comment-17640626 ] lekelei edited comment on FLINK-15656 at 11/29/22 11:18 AM: [~wangyang0918] Hello, I am using the function of pod-template, and I have a question: through the pod-template.YAML file provided in the reference document[Native Kubernetes | Apache Flink|https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/resource-providers/native_kubernetes/#pod-template], I can download the dependent jar files to the /opt/flink/artifact directory through the init container, but what is the way for the Flink jobmanager to load these jar packages? Mounting directly to the Flink jobmanager lib directory via emptyDir will overwrite the original lib directory and is not a viable method. I hope you can help me with this if you have time, thank you very much! was (Author: JIRAUSER297957): [~wangyang0918] Hello, I am using the function of pod-template, and I have a question: through the pod-template.YAML file provided in the reference document[[Native Kubernetes | Apache Flink|https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/resource-providers/native_kubernetes/#pod-template]], I can download the dependent jar files to the /opt/flink/artifact directory through the init container, but what is the way for the Flink jobmanager to load these jar packages? Mounting directly to the Flink jobmanager lib directory via emptyDir will overwrite the original lib directory and is not a viable method. I hope you can help me with this if you have time, thank you very much! > Support user-specified pod templates > > > Key: FLINK-15656 > URL: https://issues.apache.org/jira/browse/FLINK-15656 > Project: Flink > Issue Type: Sub-task > Components: Deployment / Kubernetes >Reporter: Canbin Zheng >Assignee: Yang Wang >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > > The current approach of introducing new configuration options for each aspect > of pod specification a user might wish is becoming unwieldy, we have to > maintain more and more Flink side Kubernetes configuration options and users > have to learn the gap between the declarative model used by Kubernetes and > the configuration model used by Flink. It's a great improvement to allow > users to specify pod templates as central places for all customization needs > for the jobmanager and taskmanager pods. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-15656) Support user-specified pod templates
[ https://issues.apache.org/jira/browse/FLINK-15656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640626#comment-17640626 ] lekelei commented on FLINK-15656: - [~wangyang0918] Hello, I am using the function of pod-template, and I have a question: through the pod-template.YAML file provided in the reference document[[Native Kubernetes | Apache Flink|https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/resource-providers/native_kubernetes/#pod-template]], I can download the dependent jar files to the /opt/flink/artifact directory through the init container, but what is the way for the Flink jobmanager to load these jar packages? Mounting directly to the Flink jobmanager lib directory via emptyDir will overwrite the original lib directory and is not a viable method. I hope you can help me with this if you have time, thank you very much! > Support user-specified pod templates > > > Key: FLINK-15656 > URL: https://issues.apache.org/jira/browse/FLINK-15656 > Project: Flink > Issue Type: Sub-task > Components: Deployment / Kubernetes >Reporter: Canbin Zheng >Assignee: Yang Wang >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > > The current approach of introducing new configuration options for each aspect > of pod specification a user might wish is becoming unwieldy, we have to > maintain more and more Flink side Kubernetes configuration options and users > have to learn the gap between the declarative model used by Kubernetes and > the configuration model used by Flink. It's a great improvement to allow > users to specify pod templates as central places for all customization needs > for the jobmanager and taskmanager pods. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Comment Edited] (FLINK-15656) Support user-specified pod templates
[ https://issues.apache.org/jira/browse/FLINK-15656?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640626#comment-17640626 ] lekelei edited comment on FLINK-15656 at 11/29/22 11:21 AM: [~wangyang0918] Hello, I am using the function of pod-template, and I have a question: through the pod-template.YAML file provided in the reference document [Native Kubernetes | Apache Flink |https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/resource-providers/native_kubernetes/#pod-template], I can download the dependent jar files to the /opt/flink/artifact directory through the init container, but what is the way for the Flink jobmanager to load these jar packages? Mounting directly to the Flink jobmanager lib directory via emptyDir will overwrite the original lib directory and is not a viable method. I hope you can help me with this if you have time, thank you very much! was (Author: JIRAUSER297957): [~wangyang0918] Hello, I am using the function of pod-template, and I have a question: through the pod-template.YAML file provided in the reference document[Native Kubernetes | Apache Flink|https://nightlies.apache.org/flink/flink-docs-master/zh/docs/deployment/resource-providers/native_kubernetes/#pod-template], I can download the dependent jar files to the /opt/flink/artifact directory through the init container, but what is the way for the Flink jobmanager to load these jar packages? Mounting directly to the Flink jobmanager lib directory via emptyDir will overwrite the original lib directory and is not a viable method. I hope you can help me with this if you have time, thank you very much! > Support user-specified pod templates > > > Key: FLINK-15656 > URL: https://issues.apache.org/jira/browse/FLINK-15656 > Project: Flink > Issue Type: Sub-task > Components: Deployment / Kubernetes >Reporter: Canbin Zheng >Assignee: Yang Wang >Priority: Major > Labels: pull-request-available > Fix For: 1.13.0 > > > The current approach of introducing new configuration options for each aspect > of pod specification a user might wish is becoming unwieldy, we have to > maintain more and more Flink side Kubernetes configuration options and users > have to learn the gap between the declarative model used by Kubernetes and > the configuration model used by Flink. It's a great improvement to allow > users to specify pod templates as central places for all customization needs > for the jobmanager and taskmanager pods. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink-kubernetes-operator] mbalassi commented on pull request #456: FLINK-30222: Operator should handle 'kubernetes' as a valid setting for the 'high-availability' config key
mbalassi commented on PR #456: URL: https://github.com/apache/flink-kubernetes-operator/pull/456#issuecomment-1330486354 Thanks -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi merged pull request #456: FLINK-30222: Operator should handle 'kubernetes' as a valid setting for the 'high-availability' config key
mbalassi merged PR #456: URL: https://github.com/apache/flink-kubernetes-operator/pull/456 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Closed] (FLINK-30222) Suspended a job in last-state mode bug
[
https://issues.apache.org/jira/browse/FLINK-30222?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Márton Balassi closed FLINK-30222.
--
Resolution: Fixed
[{{458022d}}|https://github.com/apache/flink-kubernetes-operator/commit/458022d2e67247c9941f102fb39d9dda96bd8837]
in main
> Suspended a job in last-state mode bug
> --
>
> Key: FLINK-30222
> URL: https://issues.apache.org/jira/browse/FLINK-30222
> Project: Flink
> Issue Type: Bug
> Components: Kubernetes Operator
>Affects Versions: 1.16.0, kubernetes-operator-1.2.0
>Reporter: tanjialiang
>Assignee: Peter Vary
>Priority: Blocker
> Labels: pull-request-available
> Fix For: kubernetes-operator-1.3.0
>
> Attachments: image-2022-11-27-16-48-08-445.png
>
>
> In flink 1.16.0, it support set kubernetes HA with options 'kubernetes', such
> as 'high-availability: kubernetes'. But in kubernetes operator 1.2.0, I try
> to suspended a job in last-state mode, it validate fail, because of 'Job
> could not be upgraded with last-state while Kubernetes HA disabled'.
>
> I try to use kubectl patch to supsended a job with last-state
> {code:sh}
> kubectl -nbigdata-flink patch
> flinkdeployments.flink.apache.org/streaming-638223bf650ac869689faa62-flink
> --type=merge -p '{"spec": {"job":
> {"state": "suspended", "upgradeMode": "last-state"}{code}
> it found an error, because my kubernetes HA is disabled
> {code:java}
> Error from server: admission webhook "flinkoperator.flink.apache.org" denied
> the request: Job could not be upgraded with last-state while Kubernetes HA
> disabled {code}
> but i enabled kubernetes HA with this follow options:
> {code:yaml}
> kubernetes.cluster-id:
> high-availability: kubernetes
> high-availability.storageDir: hdfs:///flink/recovery {code}
> and i found flink kubernetes operator 1.2.0 validate the kubernetes HA in the
> old options:
> {code:yaml}
> high-availability:
> org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
> {code}
> it may be in the
> org.apache.flink.kubernetes.operator.utils.FlinkUtils#isKubernetesHAActivated
> to judge.
> !image-2022-11-27-16-48-08-445.png!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
[GitHub] [flink] XComp merged pull request #20755: [FLINK-29092][tests] Use base path bucket assigner
XComp merged PR #20755: URL: https://github.com/apache/flink/pull/20755 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] XComp merged pull request #21413: [BP-1.16][FLINK-29092][Connectors/Test] Use one bucket assigner
XComp merged PR #21413: URL: https://github.com/apache/flink/pull/21413 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] XComp merged pull request #21414: [BP-1.15][FLINK-29092][Connectors/Test] Use one bucket assigner
XComp merged PR #21414: URL: https://github.com/apache/flink/pull/21414 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[jira] [Resolved] (FLINK-29092) HadoopPathBasedPartFileWriterTest.testWriteFile failed with AssertionError
[
https://issues.apache.org/jira/browse/FLINK-29092?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Matthias Pohl resolved FLINK-29092.
---
Fix Version/s: 1.16.1
1.15.4
Resolution: Fixed
master: ed46cb2fd64f1cb306ae5b7654d2b4d64ab69f22
1.16: 73dfd61858b7762e7d979bb8b09051abc0d82734
1.15: 84ec72f2cade11251c263a1a367634521461d225
> HadoopPathBasedPartFileWriterTest.testWriteFile failed with AssertionError
> --
>
> Key: FLINK-29092
> URL: https://issues.apache.org/jira/browse/FLINK-29092
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile)
>Affects Versions: 1.16.0, 1.17.0, 1.15.2
>Reporter: Xingbo Huang
>Assignee: KurtDing
>Priority: Major
> Labels: pull-request-available, test-stability
> Fix For: 1.17.0, 1.16.1, 1.15.4
>
> Attachments: image-2022-09-05-17-31-44-813.png
>
>
> {code:java}
> 2022-08-24T02:00:01.1670618Z Aug 24 02:00:01 [ERROR]
> org.apache.flink.formats.hadoop.bulk.HadoopPathBasedPartFileWriterTest.testWriteFile
> Time elapsed: 2.311 s <<< FAILURE!
> 2022-08-24T02:00:01.1671250Z Aug 24 02:00:01 java.lang.AssertionError:
> 2022-08-24T02:00:01.1671626Z Aug 24 02:00:01
> 2022-08-24T02:00:01.1672001Z Aug 24 02:00:01 Expected size: 1 but was: 2 in:
> 2022-08-24T02:00:01.1673656Z Aug 24 02:00:01
> [DeprecatedRawLocalFileStatus{path=file:/tmp/junit3893779198554813459/junit1595046776902782406/2022-08-24--02;
> isDirectory=true; modification_time=166130640;
> access_time=1661306400396; owner=; group=; permission=rwxrwxrwx;
> isSymlink=false; hasAcl=false; isEncrypted=false; isErasureCoded=false},
> 2022-08-24T02:00:01.1676131Z Aug 24 02:00:01
> DeprecatedRawLocalFileStatus{path=file:/tmp/junit3893779198554813459/junit1595046776902782406/2022-08-24--01;
> isDirectory=true; modification_time=166130640;
> access_time=1661306400326; owner=; group=; permission=rwxrwxrwx;
> isSymlink=false; hasAcl=false; isEncrypted=false; isErasureCoded=false}]
> 2022-08-24T02:00:01.1677339Z Aug 24 02:00:01 at
> org.apache.flink.formats.hadoop.bulk.HadoopPathBasedPartFileWriterTest.validateResult(HadoopPathBasedPartFileWriterTest.java:107)
> 2022-08-24T02:00:01.1678274Z Aug 24 02:00:01 at
> org.apache.flink.formats.hadoop.bulk.HadoopPathBasedPartFileWriterTest.testWriteFile(HadoopPathBasedPartFileWriterTest.java:97)
> 2022-08-24T02:00:01.1679017Z Aug 24 02:00:01 at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 2022-08-24T02:00:01.1679666Z Aug 24 02:00:01 at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 2022-08-24T02:00:01.1680369Z Aug 24 02:00:01 at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 2022-08-24T02:00:01.1681019Z Aug 24 02:00:01 at
> java.lang.reflect.Method.invoke(Method.java:498)
> 2022-08-24T02:00:01.1681666Z Aug 24 02:00:01 at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
> 2022-08-24T02:00:01.1682385Z Aug 24 02:00:01 at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> 2022-08-24T02:00:01.1683094Z Aug 24 02:00:01 at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
> 2022-08-24T02:00:01.1683965Z Aug 24 02:00:01 at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> 2022-08-24T02:00:01.1684713Z Aug 24 02:00:01 at
> org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
> 2022-08-24T02:00:01.1685956Z Aug 24 02:00:01 at
> org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
> 2022-08-24T02:00:01.1687185Z Aug 24 02:00:01 at
> org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
> 2022-08-24T02:00:01.1688347Z Aug 24 02:00:01 at
> java.util.concurrent.FutureTask.run(FutureTask.java:266)
> 2022-08-24T02:00:01.1689324Z Aug 24 02:00:01 at
> java.lang.Thread.run(Thread.java:748)
> 2022-08-24T02:00:01.1690062Z Aug 24 02:00:01
> 2022-08-24T02:00:03.4727706Z Aug 24 02:00:03 Formatting using clusterid:
> testClusterID
> 2022-08-24T02:00:07.9860626Z Aug 24 02:00:07 [INFO] Tests run: 12, Failures:
> 0, Errors: 0, Skipped: 0, Time elapsed: 13.712 s - in
> org.apache.flink.formats.hadoop.bulk.committer.HadoopRenameCommitterHDFSTest
> 2022-08-24T02:00:08.4139747Z Aug 24 02:00:08 [INFO]
> 2022-08-24T02:00:08.4140678Z Aug 24 02:00:08 [INFO] Results:
> 2022-08-24T02:00:08.4141326Z Aug 24 02:00:08 [INFO]
> 2022-08-24T02:00:08.4142008Z Aug 24 02:00:08 [ERROR] Failures:
> 2022-08-24T02:00:08.4144242Z Aug 24 02:00:08 [ERROR]
> HadoopPathBasedPartFileWriterTest.testWriteFile:97-
[GitHub] [flink] reswqa opened a new pull request, #21419: [FLINK-30233] Hybrid full result partition should also work with speculative execution
reswqa opened a new pull request, #21419: URL: https://github.com/apache/flink/pull/21419 ## What is the purpose of the change *Hybrid full result partition is re-consumable now, and we have enabled `VertexwiseSchedulingStrategy` to support the scheduling of hybrid type edges in [FLINK-29767](https://issues.apache.org/jira/browse/FLINK-29767). Now is the time to enable hybrid full result partition to support speculative execution. In the case of speculative execution, downstream task can be scheduled only if all upstream task finished. In the next ticket, we will introduce a new mechanism to enable the downstream to be scheduled after partial upstream task is finished, thus improving the performance.* ## Brief change log - *Speculatvie execution vertex supports mark hybrid full result partition finished.* - *Introduce AllFinishedInputConsumableDecider.* - *Speculative scheduler supports hybrid full result partition.* ## Verifying this change This change added unit tests. ## 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: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-connector-opensearch] zentol commented on a diff in pull request #1: [FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors
zentol commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-opensearch/pull/1#discussion_r1034673047
##
flink-connector-opensearch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java:
##
@@ -0,0 +1,562 @@
+/*
+ * 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.streaming.connectors.opensearch;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import
org.apache.flink.streaming.connectors.opensearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+
+import org.apache.http.HttpHost;
+import org.apache.http.HttpResponse;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.bootstrap.HttpServer;
+import org.apache.http.impl.bootstrap.ServerBootstrap;
+import org.apache.http.protocol.HttpRequestHandlerMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.DocWriteRequest.OpType;
+import org.opensearch.action.bulk.BulkItemResponse;
+import org.opensearch.action.bulk.BulkItemResponse.Failure;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Requests;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.index.shard.ShardId;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Suite of tests for {@link OpensearchSink}. */
+public class OpensearchSinkTest {
+private HttpServer server;
+private final Deque> responses = new
ConcurrentLinkedDeque<>();
+
+@BeforeEach
+public void setUp() throws IOException {
+final HttpRequestHandlerMapper handlers =
+(request) -> {
+final String method = request.getRequestLine().getMethod();
+if (method.equalsIgnoreCase("HEAD")) {
+// Connection request always OKed
+return (req, resp, context) -> resp.setStatusCode(200);
+} else if (method.equalsIgnoreCase("POST")) {
+// Bulk responses are configured per test case
+return (req, resp, context) ->
responses.poll().accept(resp);
+} else {
+return null;
+}
+};
+server =
ServerBootstrap.bootstrap().setHandlerMapper(handlers).create();
+server.start();
+}
+
+@AfterEach
+public void tearDown() {
+server.stop();
+server = null;
+responses.clear();
+}
+
+/**
+ * Tests that any item failure in the listener callbacks is rethrown on an
immediately following
+ * invoke call.
+ */
+@Test
+public void testItemFailureRethrownOnInvoke() throws Throwable {
+final OpensearchSink.Builder builder =
+new OpensearchSink.Builder<>(
+Arrays.asList(new HttpHost("localhost",
server.getLocalPort())),
+new SimpleSinkFunction());
+builder.setBulkFlushMaxActions(1);
+builder.setFailureHandler(new NoOpFailureHandler());
+
+final OpensearchSink sink = builder.build();
+final OneInputStreamOperatorT
[jira] [Updated] (FLINK-30233) Hybrid full result partition should also work with speculative execution
[ https://issues.apache.org/jira/browse/FLINK-30233?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] ASF GitHub Bot updated FLINK-30233: --- Labels: pull-request-available (was: ) > Hybrid full result partition should also work with speculative execution > > > Key: FLINK-30233 > URL: https://issues.apache.org/jira/browse/FLINK-30233 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Coordination >Affects Versions: 1.17.0 >Reporter: Weijie Guo >Assignee: Weijie Guo >Priority: Major > Labels: pull-request-available > > Hybrid full result partition is re-consumable now, and we have enabled > `VertexwiseSchedulingStrategy` to support the scheduling of hybrid type edges > in FLINK-29767. Now is the time to enable hybrid full result partition to > support speculative execution. > In the case of speculative execution, downstream task can be scheduled only > if all upstream task finished. In the next ticket, we will introduce a new > mechanism to enable the downstream to be scheduled after partial upstream > task is finished, thus improving the performance. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] flinkbot commented on pull request #21419: [FLINK-30233] Hybrid full result partition should also work with speculative execution
flinkbot commented on PR #21419: URL: https://github.com/apache/flink/pull/21419#issuecomment-1330549923 ## CI report: * 43f4095e23f3396da77a7b287d18424fd971abfe UNKNOWN Bot commands The @flinkbot bot supports the following commands: - `@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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #21418: [FLINK-30237][coordination] Only bundle single ZK client
XComp commented on code in PR #21418:
URL: https://github.com/apache/flink/pull/21418#discussion_r1034696632
##
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh:
##
@@ -113,7 +113,6 @@ function run_ha_test() {
# jm killing loop
set_config_key "env.pid.dir" "${TEST_DATA_DIR}"
-setup_flink_shaded_zookeeper ${ZOOKEEPER_VERSION}
Review Comment:
Is there a reason to keep `${ZOOKEEPER_VERSION}` as a parameter in this
method? It's not really needed anymore analogously to the other files where we
reverted mentioning the ZK version in the log output.
##
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh:
##
@@ -113,7 +113,6 @@ function run_ha_test() {
# jm killing loop
set_config_key "env.pid.dir" "${TEST_DATA_DIR}"
-setup_flink_shaded_zookeeper ${ZOOKEEPER_VERSION}
Review Comment:
We're even removing the ZK version in `run-nightly-tests.sh` as a parameter
of `test_ha_per_job_cluster_datastream.sh`. So, I guess, we can clean it up
entirely 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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[jira] [Updated] (FLINK-29769) Further limit the explosion range of failover in hybrid shuffle mode
[ https://issues.apache.org/jira/browse/FLINK-29769?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Weijie Guo updated FLINK-29769: --- Description: Under the current failover strategy, if a region changes to the failed state, all its downstream regions must be restarted. For ALL_ EDGE_BLOCKING type jobs, since they are scheduled stage by stage, no additional overhead. However, for the hybrid shuffle mode, the upstream and downstream can both run at the same time. If the upstream task fails, we hope that it will not affect the downstream tasks that do not consume it. (was: Under the current failover strategy, if a region changes to the failed state, all its downstream regions must be restarted. For ALL_ EDGE_BLOCKING type jobs, since they are scheduled stage by state, no additional overhead. However, for the hybrid shuffle mode, the upstream and downstream can both run at the same time. If the upstream task fails, we hope that it will not affect the downstream regions that do not consume it.) > Further limit the explosion range of failover in hybrid shuffle mode > > > Key: FLINK-29769 > URL: https://issues.apache.org/jira/browse/FLINK-29769 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Coordination >Affects Versions: 1.17.0 >Reporter: Weijie Guo >Priority: Major > > Under the current failover strategy, if a region changes to the failed state, > all its downstream regions must be restarted. For ALL_ EDGE_BLOCKING type > jobs, since they are scheduled stage by stage, no additional overhead. > However, for the hybrid shuffle mode, the upstream and downstream can both > run at the same time. If the upstream task fails, we hope that it will not > affect the downstream tasks that do not consume it. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Assigned] (FLINK-29769) Further limit the explosion range of failover in hybrid shuffle mode
[ https://issues.apache.org/jira/browse/FLINK-29769?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Weijie Guo reassigned FLINK-29769: -- Assignee: Weijie Guo > Further limit the explosion range of failover in hybrid shuffle mode > > > Key: FLINK-29769 > URL: https://issues.apache.org/jira/browse/FLINK-29769 > Project: Flink > Issue Type: Sub-task > Components: Runtime / Coordination >Affects Versions: 1.17.0 >Reporter: Weijie Guo >Assignee: Weijie Guo >Priority: Major > > Under the current failover strategy, if a region changes to the failed state, > all its downstream regions must be restarted. For ALL_ EDGE_BLOCKING type > jobs, since they are scheduled stage by stage, no additional overhead. > However, for the hybrid shuffle mode, the upstream and downstream can both > run at the same time. If the upstream task fails, we hope that it will not > affect the downstream tasks that do not consume it. -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21128: [FLINK-29710] Bump minimum supported Hadoop version to 2.10.2
gaborgsomogyi commented on code in PR #21128:
URL: https://github.com/apache/flink/pull/21128#discussion_r1034716206
##
flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java:
##
@@ -608,10 +611,13 @@ public boolean accept(File dir, String name) {
public static boolean verifyStringsInNamedLogFiles(
final String[] mustHave, final ApplicationId applicationId, final
String fileName) {
final List mustHaveList = Arrays.asList(mustHave);
-final File cwd = new File("target",
YARN_CONFIGURATION.get(TEST_CLUSTER_NAME_KEY));
+final File cwd =
Review Comment:
No idea, I'm just wondering how it worked until now. This area is just super
confusing in general because sometimes `target` but sometimes
`target/test/data/` is set in `YarnTestBase.java` w/o my change. At top of this
locally different execution path runs than in the CI which makes the problem
even harder to explain/debug.
What I can imagine but can't point it out:
* one `pom.xml` change made the test inherit different current directory
* YARN is getting different current directory for some magical reason
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi commented on a diff in pull request #453: [FLINK-30216] Improve code quality by standardising tab/spaces in pom…
mbalassi commented on code in PR #453: URL: https://github.com/apache/flink-kubernetes-operator/pull/453#discussion_r1034715813 ## pom.xml: ## @@ -349,6 +349,36 @@ under the License. + + +org.codehaus.mojo +xml-maven-plugin +1.0.2 + +4 +false + + +. + +** Review Comment: This seems unnecessary to me, the include is sufficient. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi commented on pull request #448: Show me every error
mbalassi commented on PR #448: URL: https://github.com/apache/flink-kubernetes-operator/pull/448#issuecomment-1330612254 @pvary is this still relevant or can this be closed? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21128: [FLINK-29710] Bump minimum supported Hadoop version to 2.10.2
gaborgsomogyi commented on code in PR #21128:
URL: https://github.com/apache/flink/pull/21128#discussion_r1034716206
##
flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java:
##
@@ -608,10 +611,13 @@ public boolean accept(File dir, String name) {
public static boolean verifyStringsInNamedLogFiles(
final String[] mustHave, final ApplicationId applicationId, final
String fileName) {
final List mustHaveList = Arrays.asList(mustHave);
-final File cwd = new File("target",
YARN_CONFIGURATION.get(TEST_CLUSTER_NAME_KEY));
+final File cwd =
Review Comment:
No idea, I'm just wondering how it worked until now. This area is just super
confusing in general because sometimes `target` but sometimes `../target` or
`target/test/data/` is set in `YarnTestBase.java` w/o my change. At top of this
locally different execution path runs than in the CI which makes the problem
even harder to explain/debug.
What I can imagine but can't point it out:
* one `pom.xml` change made the test inherit different current directory
* YARN is getting different current directory for some magical reason
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi commented on pull request #455: FLINK-30199: Add a script to run Kubernetes Operator e2e tests manually
mbalassi commented on PR #455: URL: https://github.com/apache/flink-kubernetes-operator/pull/455#issuecomment-1330618530 cc @wangyang0918 for your information @pvary has reworked the test suite to block tests together on the same container runs, we hope this effectively will give us more CI bandwidth. The minikube instances are reused between the tests, the operator itself is always reinstalled to cleanly separate the logs. I am happy to review, 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. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] gaborgsomogyi commented on a diff in pull request #21128: [FLINK-29710] Bump minimum supported Hadoop version to 2.10.2
gaborgsomogyi commented on code in PR #21128:
URL: https://github.com/apache/flink/pull/21128#discussion_r1034723930
##
flink-yarn-tests/src/test/java/org/apache/flink/yarn/YarnTestBase.java:
##
@@ -608,10 +611,13 @@ public boolean accept(File dir, String name) {
public static boolean verifyStringsInNamedLogFiles(
final String[] mustHave, final ApplicationId applicationId, final
String fileName) {
final List mustHaveList = Arrays.asList(mustHave);
-final File cwd = new File("target",
YARN_CONFIGURATION.get(TEST_CLUSTER_NAME_KEY));
+final File cwd =
Review Comment:
I think if we want to have control here then we must just forget relative
paths. Blindly all the logic which is depending on `TEST_CLUSTER_NAME_KEY` must
be just kicked out with hardcoded TMP dir providing full path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] Myasuka commented on a diff in pull request #21362: [FLINK-29430] Add sanity check when setCurrentKeyGroupIndex
Myasuka commented on code in PR #21362:
URL: https://github.com/apache/flink/pull/21362#discussion_r1034731440
##
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/InternalKeyContextImpl.java:
##
@@ -72,6 +73,10 @@ public void setCurrentKey(@Nonnull K currentKey) {
@Override
public void setCurrentKeyGroupIndex(int currentKeyGroupIndex) {
+if (!keyGroupRange.contains(currentKeyGroupIndex)) {
+throw KeyGroupRangeOffsets.newIllegalKeyGroupException(
+currentKeyGroupIndex, keyGroupRange);
+}
Review Comment:
I think Roman's suggestion is fine for developers, and for general users, as
FLINK-23908 talked about, the current check would be fine during snapshotting.
Maybe we can keep this implementation as current Zakelly did in this PR
considering the performance impact is so small.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-connector-opensearch] reta commented on a diff in pull request #1: [FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors
reta commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-opensearch/pull/1#discussion_r1034734061
##
flink-connector-opensearch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java:
##
@@ -0,0 +1,562 @@
+/*
+ * 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.streaming.connectors.opensearch;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import
org.apache.flink.streaming.connectors.opensearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+
+import org.apache.http.HttpHost;
+import org.apache.http.HttpResponse;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.bootstrap.HttpServer;
+import org.apache.http.impl.bootstrap.ServerBootstrap;
+import org.apache.http.protocol.HttpRequestHandlerMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.DocWriteRequest.OpType;
+import org.opensearch.action.bulk.BulkItemResponse;
+import org.opensearch.action.bulk.BulkItemResponse.Failure;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Requests;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.index.shard.ShardId;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Suite of tests for {@link OpensearchSink}. */
+public class OpensearchSinkTest {
+private HttpServer server;
+private final Deque> responses = new
ConcurrentLinkedDeque<>();
+
+@BeforeEach
+public void setUp() throws IOException {
+final HttpRequestHandlerMapper handlers =
+(request) -> {
+final String method = request.getRequestLine().getMethod();
+if (method.equalsIgnoreCase("HEAD")) {
+// Connection request always OKed
+return (req, resp, context) -> resp.setStatusCode(200);
+} else if (method.equalsIgnoreCase("POST")) {
+// Bulk responses are configured per test case
+return (req, resp, context) ->
responses.poll().accept(resp);
+} else {
+return null;
+}
+};
+server =
ServerBootstrap.bootstrap().setHandlerMapper(handlers).create();
+server.start();
+}
+
+@AfterEach
+public void tearDown() {
+server.stop();
+server = null;
+responses.clear();
+}
+
+/**
+ * Tests that any item failure in the listener callbacks is rethrown on an
immediately following
+ * invoke call.
+ */
+@Test
+public void testItemFailureRethrownOnInvoke() throws Throwable {
+final OpensearchSink.Builder builder =
+new OpensearchSink.Builder<>(
+Arrays.asList(new HttpHost("localhost",
server.getLocalPort())),
+new SimpleSinkFunction());
+builder.setBulkFlushMaxActions(1);
+builder.setFailureHandler(new NoOpFailureHandler());
+
+final OpensearchSink sink = builder.build();
+final OneInputStreamOperatorTes
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #451: [FLINK-30186] Bump Flink version to 1.15.3
gyfora commented on PR #451: URL: https://github.com/apache/flink-kubernetes-operator/pull/451#issuecomment-1330633259 @yangjf2019 seems like the 1.13 e2e tests started to fail on a noclassdef found error. We should probably use the respective example jars there from the 1.13 version not the latest. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-connector-opensearch] reta commented on a diff in pull request #1: [FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors
reta commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-opensearch/pull/1#discussion_r1034741956
##
flink-connector-opensearch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java:
##
@@ -0,0 +1,562 @@
+/*
+ * 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.streaming.connectors.opensearch;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import
org.apache.flink.streaming.connectors.opensearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+
+import org.apache.http.HttpHost;
+import org.apache.http.HttpResponse;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.bootstrap.HttpServer;
+import org.apache.http.impl.bootstrap.ServerBootstrap;
+import org.apache.http.protocol.HttpRequestHandlerMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.DocWriteRequest.OpType;
+import org.opensearch.action.bulk.BulkItemResponse;
+import org.opensearch.action.bulk.BulkItemResponse.Failure;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Requests;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.index.shard.ShardId;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Suite of tests for {@link OpensearchSink}. */
+public class OpensearchSinkTest {
+private HttpServer server;
+private final Deque> responses = new
ConcurrentLinkedDeque<>();
+
+@BeforeEach
+public void setUp() throws IOException {
+final HttpRequestHandlerMapper handlers =
+(request) -> {
+final String method = request.getRequestLine().getMethod();
+if (method.equalsIgnoreCase("HEAD")) {
+// Connection request always OKed
+return (req, resp, context) -> resp.setStatusCode(200);
+} else if (method.equalsIgnoreCase("POST")) {
+// Bulk responses are configured per test case
+return (req, resp, context) ->
responses.poll().accept(resp);
+} else {
+return null;
+}
+};
+server =
ServerBootstrap.bootstrap().setHandlerMapper(handlers).create();
+server.start();
+}
+
+@AfterEach
+public void tearDown() {
+server.stop();
+server = null;
+responses.clear();
+}
+
+/**
+ * Tests that any item failure in the listener callbacks is rethrown on an
immediately following
+ * invoke call.
+ */
+@Test
+public void testItemFailureRethrownOnInvoke() throws Throwable {
+final OpensearchSink.Builder builder =
+new OpensearchSink.Builder<>(
+Arrays.asList(new HttpHost("localhost",
server.getLocalPort())),
+new SimpleSinkFunction());
+builder.setBulkFlushMaxActions(1);
+builder.setFailureHandler(new NoOpFailureHandler());
+
+final OpensearchSink sink = builder.build();
+final OneInputStreamOperatorTes
[GitHub] [flink] zentol commented on a diff in pull request #21418: [FLINK-30237][coordination] Only bundle single ZK client
zentol commented on code in PR #21418:
URL: https://github.com/apache/flink/pull/21418#discussion_r1034742561
##
flink-end-to-end-tests/test-scripts/test_ha_per_job_cluster_datastream.sh:
##
@@ -113,7 +113,6 @@ function run_ha_test() {
# jm killing loop
set_config_key "env.pid.dir" "${TEST_DATA_DIR}"
-setup_flink_shaded_zookeeper ${ZOOKEEPER_VERSION}
Review Comment:
No I just missed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-connector-opensearch] reta commented on a diff in pull request #1: [FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors
reta commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-opensearch/pull/1#discussion_r1034755652
##
flink-connector-opensearch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java:
##
@@ -0,0 +1,562 @@
+/*
+ * 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.streaming.connectors.opensearch;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import
org.apache.flink.streaming.connectors.opensearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+
+import org.apache.http.HttpHost;
+import org.apache.http.HttpResponse;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.bootstrap.HttpServer;
+import org.apache.http.impl.bootstrap.ServerBootstrap;
+import org.apache.http.protocol.HttpRequestHandlerMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.DocWriteRequest.OpType;
+import org.opensearch.action.bulk.BulkItemResponse;
+import org.opensearch.action.bulk.BulkItemResponse.Failure;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Requests;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.index.shard.ShardId;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Suite of tests for {@link OpensearchSink}. */
+public class OpensearchSinkTest {
+private HttpServer server;
+private final Deque> responses = new
ConcurrentLinkedDeque<>();
+
+@BeforeEach
+public void setUp() throws IOException {
+final HttpRequestHandlerMapper handlers =
+(request) -> {
+final String method = request.getRequestLine().getMethod();
+if (method.equalsIgnoreCase("HEAD")) {
+// Connection request always OKed
+return (req, resp, context) -> resp.setStatusCode(200);
+} else if (method.equalsIgnoreCase("POST")) {
+// Bulk responses are configured per test case
+return (req, resp, context) ->
responses.poll().accept(resp);
+} else {
+return null;
+}
+};
+server =
ServerBootstrap.bootstrap().setHandlerMapper(handlers).create();
+server.start();
+}
+
+@AfterEach
+public void tearDown() {
+server.stop();
+server = null;
+responses.clear();
+}
+
+/**
+ * Tests that any item failure in the listener callbacks is rethrown on an
immediately following
+ * invoke call.
+ */
+@Test
+public void testItemFailureRethrownOnInvoke() throws Throwable {
+final OpensearchSink.Builder builder =
+new OpensearchSink.Builder<>(
+Arrays.asList(new HttpHost("localhost",
server.getLocalPort())),
+new SimpleSinkFunction());
+builder.setBulkFlushMaxActions(1);
+builder.setFailureHandler(new NoOpFailureHandler());
+
+final OpensearchSink sink = builder.build();
+final OneInputStreamOperatorTes
[jira] [Commented] (FLINK-29815) image apache/flink:1.16.0-scala_2.12 does not exist
[ https://issues.apache.org/jira/browse/FLINK-29815?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640709#comment-17640709 ] Harkin commented on FLINK-29815: Good job! The image now exists! > image apache/flink:1.16.0-scala_2.12 does not exist > --- > > Key: FLINK-29815 > URL: https://issues.apache.org/jira/browse/FLINK-29815 > Project: Flink > Issue Type: Bug >Affects Versions: 1.16.0 >Reporter: Harkin >Assignee: Márton Balassi >Priority: Minor > Attachments: image-2022-10-31-20-17-38-977.png, > image-2022-10-31-20-18-35-718.png > > > [https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/resource-providers/standalone/kubernetes/] > In the document, image `apache/flink:1.16.0-scala_2.12` was used. > !image-2022-10-31-20-17-38-977.png! > However, there is no image named `apache/flink:1.16.0-scala_2.12` in docker > hub. !image-2022-10-31-20-18-35-718.png! > [https://hub.docker.com/r/apache/flink/tags?page=1&name=1.16.0] > So, we should either modify the document or upload the image. > If the document should be modified, I'd like to make a pull request to github > repo. > -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Commented] (FLINK-30116) Don't Show Env Vars in Web UI
[ https://issues.apache.org/jira/browse/FLINK-30116?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17640713#comment-17640713 ] Chesnay Schepler commented on FLINK-30116: -- [~ConradJam] Have you made any progress? > Don't Show Env Vars in Web UI > - > > Key: FLINK-30116 > URL: https://issues.apache.org/jira/browse/FLINK-30116 > Project: Flink > Issue Type: Improvement > Components: Runtime / Web Frontend >Affects Versions: 1.16.0 >Reporter: Konstantin Knauf >Assignee: ConradJam >Priority: Blocker > Fix For: 1.17.0, 1.16.1 > > > As discussed and agreed upon in [1], we'd like to revert [2] and not show any > environment variables in the Web UI for security reasons. > [1] https://lists.apache.org/thread/rjgk15bqttvblp60zry4n5pw4xjw7q9k > [2] https://issues.apache.org/jira/browse/FLINK-28311 -- This message was sent by Atlassian Jira (v8.20.10#820010)
[GitHub] [flink] zentol commented on pull request #21416: [FLINK-30202][tests] Do not assert on checkpointId
zentol commented on PR #21416: URL: https://github.com/apache/flink/pull/21416#issuecomment-1330669367 > You're saying that we would introduce a test instability here if the RateLimitedStrategy wouldn't perform as expected? Yes. At least that was the idea. Now I'm not so sure anymore whether this makes sense. Given that the limit the count we invariably end up with `capacityPerCycle * numCycles` elements, regardless of whether rate-limiting was applied or not. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] yangjf2019 commented on pull request #451: [FLINK-30186] Bump Flink version to 1.15.3
yangjf2019 commented on PR #451: URL: https://github.com/apache/flink-kubernetes-operator/pull/451#issuecomment-1330670947 Okay,I revert it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] XComp commented on pull request #21416: [FLINK-30202][tests] Do not assert on checkpointId
XComp commented on PR #21416: URL: https://github.com/apache/flink/pull/21416#issuecomment-1330673297 yeah, that's something I was wondering as well. But the behavior of the `RateLimitedStrategy` doesn't necessarily need to be tested here, I guess. It feels like we're missing a `RateLimitedSourceReaderTest` for that kind of functionality. :thinking: -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] rgsriram commented on a diff in pull request #438: [FLINK-29974] Not allowing the cancelling the which are already in the completed state.
rgsriram commented on code in PR #438:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/438#discussion_r1034773811
##
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##
@@ -362,11 +362,22 @@ public void cancelSessionJob(
FlinkSessionJob sessionJob, UpgradeMode upgradeMode, Configuration
conf)
throws Exception {
-var jobStatus = sessionJob.getStatus().getJobStatus();
+var sessionJobStatus = sessionJob.getStatus();
+var jobStatus = sessionJobStatus.getJobStatus();
var jobIdString = jobStatus.getJobId();
Preconditions.checkNotNull(jobIdString, "The job to be suspend should
not be null");
var jobId = JobID.fromHexString(jobIdString);
Optional savepointOpt = Optional.empty();
+
+if (ReconciliationUtils.isJobInTerminalState(sessionJobStatus)) {
+LOG.info("Job is already in terminal state. JobID {}",
jobId.toHexString());
+return;
+} else if (!ReconciliationUtils.isJobRunning(sessionJobStatus)) {
Review Comment:
@gyfora - The build pipeline is failing with the below error. At this place:
https://github.com/apache/flink-kubernetes-operator/actions/runs/3574448062/jobs/6009735670.
Could you please help me here? Not sure if re-triggering solves it.
```
flinkdeployment.flink.apache.org "session-cluster-1" deleted
flinksessionjob.flink.apache.org "flink-example-statemachine" deleted
persistentvolumeclaim "session-cluster-1-pvc" deleted
ingressclass.networking.k8s.io "nginx" deleted
No resources found
Error: Process completed with exit code 1.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi commented on a diff in pull request #455: FLINK-30199: Add a script to run Kubernetes Operator e2e tests manually
mbalassi commented on code in PR #455:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/455#discussion_r1034729637
##
e2e-tests/utils.sh:
##
@@ -265,3 +270,75 @@ function create_namespace() {
fi;
}
+
+function install_cert_manager() {
+ if [[ -n ${DEBUG} ]]; then
+kubectl get pods -A
+ fi
+ kubectl apply -f
https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml
+ kubectl -n cert-manager wait --all=true --for=condition=Available
--timeout=300s deploy
+}
+
+function build_image() {
+ export SHELL=/bin/bash
+ export DOCKER_BUILDKIT=1
+ eval $(minikube docker-env)
+ docker build --progress=plain --no-cache -f ./Dockerfile -t
flink-kubernetes-operator:ci-latest --progress plain .
+ if [[ -n ${DEBUG} ]]; then
+docker images
+ fi
+}
+
+function install_operator() {
+ local namespace=$1
+
+ create_namespace flink
+ if [[ -n ${DEBUG} ]]; then
+debug="--debug"
+ fi
+ helm ${debug} install flink-kubernetes-operator -n ${namespace}
helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator
--set image.tag=ci-latest --create-namespace --set
'watchNamespaces={default,flink}'
Review Comment:
I would prefer to be able to configure the watch namespaces too for the
local testing use case, having it hardcoded might be surprising.
##
e2e-tests/run_tests.sh:
##
@@ -0,0 +1,239 @@
+#!/usr/bin/env bash
+
+# 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.
+
+
+ROOT_DIR=$(dirname $(dirname "$(readlink -f "$0")"))
+source "${ROOT_DIR}/e2e-tests/utils.sh"
Review Comment:
When running from the `e2e-tests` directory directly this results in:
```
./run_tests.sh: line 21: ./e2e-tests/utils.sh: No such file or directory
```
##
e2e-tests/utils.sh:
##
@@ -265,3 +270,75 @@ function create_namespace() {
fi;
}
+
+function install_cert_manager() {
+ if [[ -n ${DEBUG} ]]; then
+kubectl get pods -A
+ fi
+ kubectl apply -f
https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml
+ kubectl -n cert-manager wait --all=true --for=condition=Available
--timeout=300s deploy
+}
+
+function build_image() {
+ export SHELL=/bin/bash
+ export DOCKER_BUILDKIT=1
+ eval $(minikube docker-env)
+ docker build --progress=plain --no-cache -f ./Dockerfile -t
flink-kubernetes-operator:ci-latest --progress plain .
+ if [[ -n ${DEBUG} ]]; then
+docker images
+ fi
+}
+
+function install_operator() {
+ local namespace=$1
+
+ create_namespace flink
+ if [[ -n ${DEBUG} ]]; then
+debug="--debug"
+ fi
+ helm ${debug} install flink-kubernetes-operator -n ${namespace}
helm/flink-kubernetes-operator --set image.repository=flink-kubernetes-operator
--set image.tag=ci-latest --create-namespace --set
'watchNamespaces={default,flink}'
Review Comment:
@pvary on second thought it is up to you how far you want to take this as
this would mean also making the CR namespaces customizable, your call whether
you think it is worth it.
##
e2e-tests/run_tests.sh:
##
@@ -0,0 +1,239 @@
+#!/usr/bin/env bash
+
+# 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.
+###
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #451: [FLINK-30186] Bump Flink version to 1.15.3
gyfora commented on PR #451: URL: https://github.com/apache/flink-kubernetes-operator/pull/451#issuecomment-1330682062 @yangjf2019 do you think we could modify the logic so that we pick the jars based on Flink version, similar to how change other parts of the spec? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi commented on pull request #447: DO_NOT_MERGE [FLINK-30151] Remove AuditUtils from operator error
mbalassi commented on PR #447: URL: https://github.com/apache/flink-kubernetes-operator/pull/447#issuecomment-1330682449 @gaborgsomogyi is this still relevant or can we close this? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] darenwkt commented on a diff in pull request #453: [FLINK-30216] Improve code quality by standardising tab/spaces in pom…
darenwkt commented on code in PR #453: URL: https://github.com/apache/flink-kubernetes-operator/pull/453#discussion_r1034783258 ## pom.xml: ## @@ -349,6 +349,36 @@ under the License. + + +org.codehaus.mojo +xml-maven-plugin +1.0.2 + +4 +false + + +. + +** Review Comment: Hi @mbalassi, thank you for the feedback, I agree with your commend that include alone is sufficient. Have updated the PR, thank you -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] gaborgsomogyi commented on pull request #447: DO_NOT_MERGE [FLINK-30151] Remove AuditUtils from operator error
gaborgsomogyi commented on PR #447: URL: https://github.com/apache/flink-kubernetes-operator/pull/447#issuecomment-1330690877 At some point I need to come back and execute all tests to find out the reason. Now we've added several bugfixes so I need to rebase. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-connector-opensearch] reta commented on a diff in pull request #1: [FLINK-25756] [connectors/opensearch] Dedicated Opensearch connectors
reta commented on code in PR #1:
URL:
https://github.com/apache/flink-connector-opensearch/pull/1#discussion_r1034741956
##
flink-connector-opensearch/src/test/java/org/apache/flink/streaming/connectors/opensearch/OpensearchSinkTest.java:
##
@@ -0,0 +1,562 @@
+/*
+ * 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.streaming.connectors.opensearch;
+
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.CheckedThread;
+import org.apache.flink.streaming.api.operators.StreamSink;
+import
org.apache.flink.streaming.connectors.opensearch.util.NoOpFailureHandler;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.MockStreamingRuntimeContext;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+
+import org.apache.http.HttpHost;
+import org.apache.http.HttpResponse;
+import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.bootstrap.HttpServer;
+import org.apache.http.impl.bootstrap.ServerBootstrap;
+import org.apache.http.protocol.HttpRequestHandlerMapper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.opensearch.action.ActionRequest;
+import org.opensearch.action.DocWriteRequest.OpType;
+import org.opensearch.action.bulk.BulkItemResponse;
+import org.opensearch.action.bulk.BulkItemResponse.Failure;
+import org.opensearch.action.bulk.BulkResponse;
+import org.opensearch.action.index.IndexResponse;
+import org.opensearch.client.Requests;
+import org.opensearch.common.xcontent.ToXContent;
+import org.opensearch.common.xcontent.XContentBuilder;
+import org.opensearch.common.xcontent.json.JsonXContent;
+import org.opensearch.index.shard.ShardId;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Consumer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Suite of tests for {@link OpensearchSink}. */
+public class OpensearchSinkTest {
+private HttpServer server;
+private final Deque> responses = new
ConcurrentLinkedDeque<>();
+
+@BeforeEach
+public void setUp() throws IOException {
+final HttpRequestHandlerMapper handlers =
+(request) -> {
+final String method = request.getRequestLine().getMethod();
+if (method.equalsIgnoreCase("HEAD")) {
+// Connection request always OKed
+return (req, resp, context) -> resp.setStatusCode(200);
+} else if (method.equalsIgnoreCase("POST")) {
+// Bulk responses are configured per test case
+return (req, resp, context) ->
responses.poll().accept(resp);
+} else {
+return null;
+}
+};
+server =
ServerBootstrap.bootstrap().setHandlerMapper(handlers).create();
+server.start();
+}
+
+@AfterEach
+public void tearDown() {
+server.stop();
+server = null;
+responses.clear();
+}
+
+/**
+ * Tests that any item failure in the listener callbacks is rethrown on an
immediately following
+ * invoke call.
+ */
+@Test
+public void testItemFailureRethrownOnInvoke() throws Throwable {
+final OpensearchSink.Builder builder =
+new OpensearchSink.Builder<>(
+Arrays.asList(new HttpHost("localhost",
server.getLocalPort())),
+new SimpleSinkFunction());
+builder.setBulkFlushMaxActions(1);
+builder.setFailureHandler(new NoOpFailureHandler());
+
+final OpensearchSink sink = builder.build();
+final OneInputStreamOperatorTes
[GitHub] [flink] zentol commented on pull request #21416: [FLINK-30202][tests] Do not assert on checkpointId
zentol commented on PR #21416: URL: https://github.com/apache/flink/pull/21416#issuecomment-1330691876 There is a `RateLimitedSourceReaderITCase`. I'll try finding another way to test this; my current thinking goes towards using a FlatMapFunction that stops emitting values after the first call to `snapshotState`, so it _should_ truly only emit the values of the first checkpoint. but that's so far also not working; to many values get emitted... -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] mbalassi commented on a diff in pull request #453: [FLINK-30216] Improve code quality by standardising tab/spaces in pom…
mbalassi commented on code in PR #453: URL: https://github.com/apache/flink-kubernetes-operator/pull/453#discussion_r1034789046 ## pom.xml: ## @@ -349,6 +349,36 @@ under the License. + + +org.codehaus.mojo +xml-maven-plugin +1.0.2 + +4 +false + + +. + +** Review Comment: Roger that. I will merge as soon as the first build returns green. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
[GitHub] [flink] XComp commented on a diff in pull request #20919: [FLINK-29405] Fix unstable test InputFormatCacheLoaderTest
XComp commented on code in PR #20919:
URL: https://github.com/apache/flink/pull/20919#discussion_r1034405416
##
flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/functions/table/fullcache/inputformat/InputFormatCacheLoaderTest.java:
##
@@ -149,19 +155,33 @@ void testCloseAndInterruptDuringReload() throws Exception
{
Future future = executorService.submit(cacheLoader);
executorService.shutdownNow(); // internally interrupts a thread
assertThatNoException().isThrownBy(future::get); // wait for the end
-// check that we didn't process all elements, but reacted on
interruption
-assertThat(sleepCounter).hasValueLessThan(totalSleepCount);
assertThat(metricGroup.numLoadFailuresCounter.getCount()).isEqualTo(0);
+}
-sleepCounter.set(0);
+@Test
+void testCloseDuringReload() throws Exception {
+AtomicInteger recordsCounter = new AtomicInteger(0);
+int totalRecords = TestCacheLoader.DATA.size() + 1; // 1 key with 2
records
+CountDownLatch latch = new CountDownLatch(1);
+Runnable reloadAction =
+ThrowingRunnable.unchecked(
+() -> {
+recordsCounter.incrementAndGet();
+latch.await();
+});
+InputFormatCacheLoader cacheLoader = createCacheLoader(0,
reloadAction);
+InterceptingCacheMetricGroup metricGroup = new
InterceptingCacheMetricGroup();
+cacheLoader.open(metricGroup);
// check closing
-executorService = Executors.newSingleThreadExecutor();
-future = executorService.submit(cacheLoader);
+ExecutorService executorService = Executors.newSingleThreadExecutor();
+Future future = executorService.submit(cacheLoader);
cacheLoader.close();
-assertThatNoException().isThrownBy(future::get); // wait for the end
+latch.countDown();
+future.get(); // wait for the end
+executorService.shutdown();
Review Comment:
you're not waiting for the shutdown to terminate here.
##
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/table/lookup/fullcache/inputformat/InputFormatCacheLoader.java:
##
@@ -107,7 +109,11 @@ protected void reloadCache() throws Exception {
} catch (InterruptedException ignored) { // we use interrupt to close
reload thread
} finally {
if (cacheLoadTaskService != null) {
+// if main cache reload thread encountered an exception,
+// it interrupts underlying InputSplitCacheLoadTasks threads
cacheLoadTaskService.shutdownNow();
Review Comment:
No, I struggle to come up with a better approach. @zentol any thoughts from
you on that matter. :thinking:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink-kubernetes-operator] yangjf2019 commented on pull request #451: [FLINK-30186] Bump Flink version to 1.15.3
yangjf2019 commented on PR #451: URL: https://github.com/apache/flink-kubernetes-operator/pull/451#issuecomment-1330698087 Yes, I think it can, in that case, how should we adjust it please? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
