You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/09/05 21:40:46 UTC
[GitHub] [flink] AHeise opened a new pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
AHeise opened a new pull request #17152:
URL: https://github.com/apache/flink/pull/17152
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Producer can leak if there are interruptions during cancellation while the producer tries to orderly abort transactions.
## Brief change log
- Harden KafkaSinkITCase and active FLIP-147
- Fix committer bugs with active FLIP-147 (duplicate `SinkWriter#prepareCommit`)
- Make closing of committer and Kafka writer/committer more reliable by not failing on first exception
- 3 smaller fixes in KafkaWriter/Committer.
## Verifying this change
Added test with concurrent checkpoints to KafkaWriterITCase.
Most fixes cover issues exposed by existing tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702838686
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -155,7 +157,7 @@
}
this.kafkaWriterState = new KafkaWriterState(transactionalIdPrefix);
- this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(-1);
+ this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(0);
Review comment:
I added it to `CheckpointIDCounter` which is non-public, but we should probably expose it some public class if we intend to commit to default = 1 (which I guess quite a few users rely on).
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702814122
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
##########
@@ -350,4 +365,17 @@ private static void transitionTransactionManagerStateTo(
Object transactionManager, String state) {
invoke(transactionManager, "transitionTo", getTransactionManagerState(state));
}
+
+ @Override
+ public String toString() {
+ return "FlinkKafkaInternalProducer{"
+ + "transactionalId='"
+ + transactionalId
+ + '\''
+ + ", inTransaction="
Review comment:
Strange - it's autogenerated.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702814948
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
Review comment:
This was an unintended change but now I intend to keep it. I don't think we can and should handle Errors here (think of OOM).
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913540205
FLINK-23850 job:
```
Configuration config = new Configuration();
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(20, 2000));
env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
env.setParallelism(6);
final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
tableEnv.createTable("T1",
TableDescriptor.forConnector("kafka")
.schema(Schema.newBuilder()
.column("pk", DataTypes.STRING().notNull())
.column("x", DataTypes.STRING().notNull())
.build())
.option("topic", "flink-23850-in1")
.option("properties.bootstrap.servers", FLINK23850Utils.BOOTSTRAP_SERVERS)
.option("value.format", "csv")
.option("scan.startup.mode", "earliest-offset")
.build());
final Table resultTable =
tableEnv.sqlQuery(
"SELECT "
+ "T1.pk, "
+ "'asd', "
+ "'foo', "
+ "'bar' "
+ "FROM T1");
tableEnv.createTable("T4",
TableDescriptor.forConnector("kafka")
.schema(Schema.newBuilder()
.column("pk", DataTypes.STRING().notNull())
.column("some_calculated_value", DataTypes.STRING())
.column("pk1", DataTypes.STRING())
.column("pk2", DataTypes.STRING())
.build())
.option("topic", "flink-23850-out")
.option("properties.bootstrap.servers", FLINK23850Utils.BOOTSTRAP_SERVERS)
.option("value.format", "csv")
.option("sink.delivery-guarantee", "exactly-once")
.option("sink.transactional-id-prefix", "flink-23850")
.option("scan.startup.mode", "earliest-offset")
.build());
resultTable.executeInsert("T4");
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
}, {
"hash" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632",
"triggerID" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"triggerType" : "PUSH"
}, {
"hash" : "f9ce8aae722f6bed3bc6e47052479caf557570ac",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23648",
"triggerID" : "f9ce8aae722f6bed3bc6e47052479caf557570ac",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f9ce8aae722f6bed3bc6e47052479caf557570ac Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23648)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702927674
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -125,12 +122,17 @@
this.kafkaProducerConfig = checkNotNull(kafkaProducerConfig, "kafkaProducerConfig");
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
this.recordSerializer = checkNotNull(recordSerializer, "recordSerializer");
+ MailboxExecutor mailboxExecutor = sinkInitContext.getMailboxExecutor();
this.deliveryCallback =
(metadata, exception) -> {
- if (exception != null && producerAsyncException == null) {
- producerAsyncException = exception;
+ if (exception != null) {
+ mailboxExecutor.execute(
Review comment:
Thanks for the clarification
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
}, {
"hash" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632",
"triggerID" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 640fa7cee84eccde4c60157cfbc3d18e2d3a618f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702839096
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -125,12 +122,17 @@
this.kafkaProducerConfig = checkNotNull(kafkaProducerConfig, "kafkaProducerConfig");
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
this.recordSerializer = checkNotNull(recordSerializer, "recordSerializer");
+ MailboxExecutor mailboxExecutor = sinkInitContext.getMailboxExecutor();
this.deliveryCallback =
(metadata, exception) -> {
- if (exception != null && producerAsyncException == null) {
- producerAsyncException = exception;
+ if (exception != null) {
+ mailboxExecutor.execute(
Review comment:
This throws the exception in the main task thread leading to regular failover. The callback is potentially in a different thread.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913537258
Running the FLINK-23850 job on this PR's codebase results in the following error with limited data being added to the Kafka input topic:
```
2021-09-06 13:07:45
java.lang.RuntimeException: Failed to send data to Kafka
at org.apache.flink.connector.kafka.sink.KafkaWriter.lambda$null$0(KafkaWriter.java:131)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:338)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:324)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:789)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:741)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.UnknownProducerIdException: This exception is raised by the broker if it could not locate the producer metadata associated with the producerId in question. This could happen if, for instance, the producer's records were deleted because their retention time had elapsed. Once the last records of the producerId are removed, the producer's metadata is removed from the broker, and future appends by the producer will return this exception.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp removed a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp removed a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913540963
A second run on the same cluster with more data caused the following Exceptions:
```
org.apache.flink.kafka.shaded.org.apache.kafka.common.KafkaException: Cannot perform send because at least one previous transactional or idempotent request has failed with errors.
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.internals.TransactionManager.failIfNotReadyForSend(TransactionManager.java:356)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:926)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:865)
at org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:184)
at org.apache.flink.streaming.runtime.operators.sink.SinkOperator.processElement(SinkOperator.java:166)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
at StreamExecCalc$6.processElement(Unknown Source)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:196)
at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:36)
at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:27)
at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:141)
at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:341)
at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:490)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:789)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:741)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.InvalidRecordException: This record has failed the validation on broker and hence be rejected.
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddc241c3ce922b83edff8cd73fd3215ec5eda330 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
}, {
"hash" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4572e70e68e47877631d12907a767343845afa98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620)
* 640fa7cee84eccde4c60157cfbc3d18e2d3a618f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702928951
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -189,7 +191,7 @@ public void write(IN element, Context context) throws IOException {
final List<KafkaCommittable> committables =
Collections.singletonList(
KafkaCommittable.of(currentProducer, producerPool::add));
- LOG.info("Committing {} committables.", committables);
+ LOG.debug("Committing {} committables {}.", committables, flush);
Review comment:
Shouldn't we pass in `deliveryGuarantee != DeliveryGuarantee.NONE || flush`. The `flush` is not only called in case `flush == true`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp removed a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp removed a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913554636
Here is a tar archive containing the logs of both runs and the Flink clusters configuration.
[FLINK-24131.tar.gz](https://github.com/apache/flink/files/7115424/FLINK-24131.tar.gz)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577)
* 9e722ba86feb67261c9a49a59d74ffea620b2aa5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613)
* 4572e70e68e47877631d12907a767343845afa98 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9e722ba86feb67261c9a49a59d74ffea620b2aa5 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613)
* 4572e70e68e47877631d12907a767343845afa98 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702833664
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -155,7 +157,7 @@
}
this.kafkaWriterState = new KafkaWriterState(transactionalIdPrefix);
- this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(-1);
+ this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(0);
Review comment:
It's a good idea to introduce a constant but `CheckpointConfig` is probably not the correct place.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702843536
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
##########
@@ -72,6 +73,7 @@ public CommitterOperator(
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
committerHandler.initializeState(context);
+ commitRetrier.retryWithDelay();
Review comment:
Added a line but commit message should be pretty clear.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913537258
Running the FLINK-23850 job on this PR's codebase results in the following error with limited data being added to the Kafka input topic:
```
java.lang.RuntimeException: Failed to send data to Kafka
at org.apache.flink.connector.kafka.sink.KafkaWriter.lambda$null$0(KafkaWriter.java:131)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:338)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:324)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:789)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:741)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.InvalidRecordException: This record has failed the validation on broker and hence be rejected.
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913540963
A second run on the same cluster with more data caused the following Exceptions:
```
org.apache.flink.kafka.shaded.org.apache.kafka.common.KafkaException: Cannot perform send because at least one previous transactional or idempotent request has failed with errors.
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.internals.TransactionManager.failIfNotReadyForSend(TransactionManager.java:356)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:926)
at org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:865)
at org.apache.flink.connector.kafka.sink.KafkaWriter.write(KafkaWriter.java:184)
at org.apache.flink.streaming.runtime.operators.sink.SinkOperator.processElement(SinkOperator.java:166)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
at org.apache.flink.streaming.api.operators.StreamFilter.processElement(StreamFilter.java:39)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:56)
at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:29)
at StreamExecCalc$6.processElement(Unknown Source)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:82)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:57)
at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:29)
at org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask$AsyncDataOutputToOutput.emitRecord(SourceOperatorStreamTask.java:196)
at org.apache.flink.streaming.api.operators.source.SourceOutputWithWatermarks.collect(SourceOutputWithWatermarks.java:110)
at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:36)
at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:27)
at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:141)
at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:341)
at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:490)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:789)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:741)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.InvalidRecordException: This record has failed the validation on broker and hence be rejected.
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702934683
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
+ LOG.error(
+ "Transaction {} encountered error and data has been potentially lost.",
+ committable,
+ e);
}
+ recyclable.ifPresent(Recyclable::close);
Review comment:
True. IMHO, the code is hard to understand here because of the little `continue` being hidden in the catch blocks. But I cannot come up with a quick fix. ¯\_(ツ)_/¯
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702888033
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
##########
@@ -197,12 +212,28 @@ public void testRecoveryWithAtLeastOnceGuarantee() throws Exception {
public void testRecoveryWithExactlyOnceGuarantee() throws Exception {
testRecoveryWithAssertion(
DeliveryGuarantee.EXACTLY_ONCE,
+ 1,
+ (records) ->
+ assertThat(
+ records,
+ contains(
+ LongStream.range(1, lastCheckpointedRecord.get().get() + 1)
+ .boxed()
+ .toArray())));
+ }
+
+ @Test
+ public void testRecoveryWithExactlyOnceGuaranteeAndConcurrentCheckpoints() throws Exception {
Review comment:
Yes, I ran all tests 100 times before and it didn't fail once. You just have different timings on a completely overloaded AZP.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702845196
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
+ LOG.error(
+ "Transaction {} encountered error and data has been potentially lost.",
+ committable,
+ e);
}
+ recyclable.ifPresent(Recyclable::close);
Review comment:
> Why can we close the producer if we might want to retry committing the Committables? 🤔
We are not retrying them at this point. Retriable path is short-cutted with `continue`.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
+ LOG.error(
+ "Transaction {} encountered error and data has been potentially lost.",
+ committable,
+ e);
}
+ recyclable.ifPresent(Recyclable::close);
Review comment:
> Why can we close the producer if we might want to retry committing the Committables? 🤔
We are not retrying them at this point. Retriable path is short-cutted with `continue`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702936188
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
##########
@@ -123,6 +130,13 @@
private SharedReference<AtomicBoolean> failed;
private SharedReference<AtomicLong> lastCheckpointedRecord;
+ @Parameterized.Parameter public int run;
+
+ @Parameterized.Parameters
+ public static Set<Integer> getConfigurations() {
Review comment:
fair point 👍
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddc241c3ce922b83edff8cd73fd3215ec5eda330 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562)
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913537258
Running the FLINK-23850 job on this PR's codebase results in the following error with constant data being added to the Kafka input topic:
```
2021-09-06 13:07:45
java.lang.RuntimeException: Failed to send data to Kafka
at org.apache.flink.connector.kafka.sink.KafkaWriter.lambda$null$0(KafkaWriter.java:131)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:338)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:324)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:789)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:741)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.UnknownProducerIdException: This exception is raised by the broker if it could not locate the producer metadata associated with the producerId in question. This could happen if, for instance, the producer's records were deleted because their retention time had elapsed. Once the last records of the producerId are removed, the producer's metadata is removed from the broker, and future appends by the producer will return this exception.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913564795
Here is a tar archive containing the `DEBUG` logs of both runs and the Flink clusters configuration.
[FLINK-24131-2.tar.gz](https://github.com/apache/flink/files/7115563/FLINK-24131-2.tar.gz)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913540205
FLINK-23850 job:
```
Configuration config = new Configuration();
config.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, true);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(config);
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(20, 2000));
env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
env.setParallelism(6);
final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
tableEnv.createTable("T1",
TableDescriptor.forConnector("kafka")
.schema(Schema.newBuilder()
.column("pk", DataTypes.STRING().notNull())
.column("x", DataTypes.STRING().notNull())
.build())
.option("topic", "flink-23850-in1")
.option("properties.bootstrap.servers", FLINK23850Utils.BOOTSTRAP_SERVERS)
.option("value.format", "csv")
.option("scan.startup.mode", "earliest-offset")
.build());
final Table resultTable =
tableEnv.sqlQuery(
"SELECT "
+ "T1.pk, "
+ "'asd', "
+ "'foo', "
+ "'bar' "
+ "FROM T1");
tableEnv.createTable("T4",
TableDescriptor.forConnector("kafka")
.schema(Schema.newBuilder()
.column("pk", DataTypes.STRING().notNull())
.column("some_calculated_value", DataTypes.STRING())
.column("pk1", DataTypes.STRING())
.column("pk2", DataTypes.STRING())
.build())
.option("topic", "flink-23850-out")
.option("properties.bootstrap.servers", FLINK23850Utils.BOOTSTRAP_SERVERS)
.option("value.format", "csv")
.option("sink.delivery-guarantee", "exactly-once")
.option("sink.transactional-id-prefix", "flink-23850")
.option("scan.startup.mode", "earliest-offset")
.build());
resultTable.executeInsert("T4");
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913554636
Here is a tar archive containing the logs of both runs and the Flink clusters configuration.
[FLINK-24131.tar.gz](https://github.com/apache/flink/files/7115424/FLINK-24131.tar.gz)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise merged pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise merged pull request #17152:
URL: https://github.com/apache/flink/pull/17152
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
}, {
"hash" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632",
"triggerID" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"triggerType" : "PUSH"
}, {
"hash" : "f9ce8aae722f6bed3bc6e47052479caf557570ac",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23648",
"triggerID" : "f9ce8aae722f6bed3bc6e47052479caf557570ac",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 640fa7cee84eccde4c60157cfbc3d18e2d3a618f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632)
* f9ce8aae722f6bed3bc6e47052479caf557570ac Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23648)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702932785
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
##########
@@ -72,6 +73,7 @@ public CommitterOperator(
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
committerHandler.initializeState(context);
+ commitRetrier.retryWithDelay();
Review comment:
"as quickly as possible" kind-of contradicts "retryWithDelay" ;-) Would it make sense to extend the public API of `CommitRetrier` to provide a method `retry` method without any delay? 🤔
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577)
* 9e722ba86feb67261c9a49a59d74ffea620b2aa5 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddc241c3ce922b83edff8cd73fd3215ec5eda330 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
}, {
"hash" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632",
"triggerID" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"triggerType" : "PUSH"
}, {
"hash" : "f9ce8aae722f6bed3bc6e47052479caf557570ac",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f9ce8aae722f6bed3bc6e47052479caf557570ac",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 640fa7cee84eccde4c60157cfbc3d18e2d3a618f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632)
* f9ce8aae722f6bed3bc6e47052479caf557570ac UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913564795
Here is a tar archive containing the logs of both runs and the Flink clusters configuration.
[Uploading FLINK-24131.tar.gz…]()
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913232537
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit ddc241c3ce922b83edff8cd73fd3215ec5eda330 (Sun Sep 05 21:44:18 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702662226
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/FlinkKafkaInternalProducer.java
##########
@@ -350,4 +365,17 @@ private static void transitionTransactionManagerStateTo(
Object transactionManager, String state) {
invoke(transactionManager, "transitionTo", getTransactionManagerState(state));
}
+
+ @Override
+ public String toString() {
+ return "FlinkKafkaInternalProducer{"
+ + "transactionalId='"
+ + transactionalId
+ + '\''
+ + ", inTransaction="
Review comment:
```suggestion
+ "', inTransaction="
```
nit
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
Review comment:
```suggestion
"Transaction {} timed out or was overridden and data has been potentially lost.",
```
nit
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -125,12 +122,17 @@
this.kafkaProducerConfig = checkNotNull(kafkaProducerConfig, "kafkaProducerConfig");
this.transactionalIdPrefix = checkNotNull(transactionalIdPrefix, "transactionalIdPrefix");
this.recordSerializer = checkNotNull(recordSerializer, "recordSerializer");
+ MailboxExecutor mailboxExecutor = sinkInitContext.getMailboxExecutor();
this.deliveryCallback =
(metadata, exception) -> {
- if (exception != null && producerAsyncException == null) {
- producerAsyncException = exception;
+ if (exception != null) {
+ mailboxExecutor.execute(
Review comment:
Can you shortly explain what throwing an `RuntimeException` on the mailbox executor would trigger?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -325,18 +318,17 @@ private void flushRecords(boolean finalFlush) {
producer = new FlinkKafkaInternalProducer<>(kafkaProducerConfig, transactionalId);
closer.register(producer);
producer.initTransactions();
- initMetrics(producer);
+ initKafkaMetrics(producer);
} else {
producer.initTransactionId(transactionalId);
}
return producer;
}
- private void initMetrics(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
+ private void initKafkaMetrics(FlinkKafkaInternalProducer<byte[], byte[]> producer) {
byteOutMetric =
MetricUtil.getKafkaMetric(
producer.metrics(), "producer-metrics", "outgoing-byte-total");
Review comment:
Should we move the metrics labels into constants? At least, `producer-metrics` is used multiple times within `KafkaWriter`.
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaCommitterTest.java
##########
@@ -31,12 +34,37 @@
/** Tests for {@link KafkaCommitter}. */
public class KafkaCommitterTest extends TestLogger {
+ private static final int PRODUCER_ID = 0;
+ private static final short EPOCH = 0;
+ private static final String TRANSACTIONAL_ID = "transactionalId";
+
+ /** Causes a network error by inactive broker and tests that a retry will happen. */
@Test
- public void testRetryCommittableOnFailure() throws IOException {
- final KafkaCommitter committer = new KafkaCommitter(new Properties());
- final short epoch = 0;
+ public void testRetryCommittableOnRetriableError() throws IOException {
+ final KafkaCommitter committer = new KafkaCommitter(getProperties());
final List<KafkaCommittable> committables =
- Collections.singletonList(new KafkaCommittable(0, epoch, "transactionalId", null));
+ Collections.singletonList(
+ new KafkaCommittable(PRODUCER_ID, EPOCH, TRANSACTIONAL_ID, null));
assertEquals(committables, committer.commit(committables));
}
+
+ /** Causes a permanent error by misconfiguration. */
+ @Test
+ public void testRetryCommittableOnFatalError() throws IOException {
+ final KafkaCommitter committer = new KafkaCommitter(new Properties());
Review comment:
```suggestion
@Test
public void testRetryCommittableOnFatalError() throws IOException {
// Causes a permanent error by misconfiguration.
final KafkaCommitter committer = new KafkaCommitter(new Properties());
```
nit: the comment actually describes the behaviour caused by the `new Properties()` parameter value. It doesn't describe the test itself which is totally fine because of the test method being descriptive enough.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -206,14 +208,26 @@ public void write(IN element, Context context) throws IOException {
@Override
public void close() throws Exception {
+ closed = true;
+ closeAll(
+ this::abortCurrentProducer,
+ closer,
+ producerPool::clear,
+ () -> {
+ checkState(currentProducer.isClosed());
+ currentProducer = null;
+ });
+ }
+
+ private void abortCurrentProducer() {
if (currentProducer.isInTransaction()) {
- currentProducer.abortTransaction();
+ try {
+ currentProducer.abortTransaction();
+ } catch (ProducerFencedException e) {
+ LOG.trace(
Review comment:
Does this happen that often that we shouldn't use `debug` here?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
##########
@@ -72,6 +73,7 @@ public CommitterOperator(
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
committerHandler.initializeState(context);
+ commitRetrier.retryWithDelay();
Review comment:
Why is it necessary to retry the handler during initialization of the operator? To me as a reader, it isn't that obviously clear. Would it make sense to add a comment here?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
+ LOG.error(
+ "Transaction {} encountered error and data has been potentially lost.",
+ committable,
+ e);
}
+ recyclable.ifPresent(Recyclable::close);
Review comment:
Why can we close the producer if we might want to retry committing the Committables? 🤔
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
##########
@@ -197,12 +212,28 @@ public void testRecoveryWithAtLeastOnceGuarantee() throws Exception {
public void testRecoveryWithExactlyOnceGuarantee() throws Exception {
testRecoveryWithAssertion(
DeliveryGuarantee.EXACTLY_ONCE,
+ 1,
+ (records) ->
+ assertThat(
+ records,
+ contains(
+ LongStream.range(1, lastCheckpointedRecord.get().get() + 1)
+ .boxed()
+ .toArray())));
+ }
+
+ @Test
+ public void testRecoveryWithExactlyOnceGuaranteeAndConcurrentCheckpoints() throws Exception {
Review comment:
I would have expected this test to fail without your changes considering that we ran into issues when doing the release testing in [FLINK-23850](https://issues.apache.org/jira/browse/FLINK-23850) resulting in [FLINK-24151](https://issues.apache.org/jira/browse/FLINK-24151). But reverting all you changes and only applying the `KafkaSinkITCase` diff didn't result in any failure on my local machine after 80 runs of `testRecoveryWithExactlyOnceGuaranteeAndConcurrentCheckpoints`
Is this expected? 🤔
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaCommitterTest.java
##########
@@ -31,12 +34,37 @@
/** Tests for {@link KafkaCommitter}. */
public class KafkaCommitterTest extends TestLogger {
+ private static final int PRODUCER_ID = 0;
+ private static final short EPOCH = 0;
+ private static final String TRANSACTIONAL_ID = "transactionalId";
+
+ /** Causes a network error by inactive broker and tests that a retry will happen. */
@Test
- public void testRetryCommittableOnFailure() throws IOException {
- final KafkaCommitter committer = new KafkaCommitter(new Properties());
- final short epoch = 0;
+ public void testRetryCommittableOnRetriableError() throws IOException {
+ final KafkaCommitter committer = new KafkaCommitter(getProperties());
Review comment:
```suggestion
// causes a network error by inactive broker
final KafkaCommitter committer = new KafkaCommitter(getProperties());
```
nit: I would remove the test method comment. The method name is descriptive enough to show that we're testing the retry here. Why the configuration returned by `getProperties()` causes the expected behavior is valuable, though, and should be added to the respective line.
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -155,7 +157,7 @@
}
this.kafkaWriterState = new KafkaWriterState(transactionalIdPrefix);
- this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(-1);
+ this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(0);
Review comment:
Or should we use `CheckpointConfig.DEFAULT_CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA` instead? 🤔
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
+ LOG.error(
+ "Transaction {} encountered error and data has been potentially lost.",
+ committable,
+ e);
}
+ recyclable.ifPresent(Recyclable::close);
Review comment:
Could we add a test for the closing to` KafkaCommitterTest`?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -165,20 +167,20 @@
|| deliveryGuarantee == DeliveryGuarantee.NONE) {
this.currentProducer = new FlinkKafkaInternalProducer<>(this.kafkaProducerConfig, null);
closer.register(this.currentProducer);
- initMetrics(this.currentProducer);
+ initKafkaMetrics(this.currentProducer);
} else {
throw new UnsupportedOperationException(
"Unsupported Kafka writer semantic " + this.deliveryGuarantee);
}
+
+ metricGroup.setCurrentSendTimeGauge(() -> computeSendTime());
Review comment:
Would it make sense to have a separate method `initFlinkMetrics` to cover that line analogously to `initKafkaMetrics`?
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
##########
@@ -123,6 +130,13 @@
private SharedReference<AtomicBoolean> failed;
private SharedReference<AtomicLong> lastCheckpointedRecord;
+ @Parameterized.Parameter public int run;
+
+ @Parameterized.Parameters
+ public static Set<Integer> getConfigurations() {
Review comment:
Would it make sense to add some comment here describing the intention of `run`? `run` is never used in any of the methods. I guess, the intention is that we want to loop over the tests for some reason? `getConfigurations` is a quite generic term to describe the intention. Maybe, there's a better name (like `getTestRunCounter()`) to describe what this method returns?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -155,7 +157,7 @@
}
this.kafkaWriterState = new KafkaWriterState(transactionalIdPrefix);
- this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(-1);
+ this.lastCheckpointId = sinkInitContext.getRestoredCheckpointId().orElse(0);
Review comment:
We change it to `0` because the checkpoint ID counting starts at `1`. That knowledge was a bit hidden in [StandaloneCheckpointIDCounter:34](https://github.com/apache/flink/blob/c6997c97c575d334679915c328792b8a3067cfb5/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/StandaloneCheckpointIDCounter.java#L34), [ZooKeeperCheckpointIDCounter:86](https://github.com/apache/flink/blob/79936be37dff2756f3829f89deec00a676db323d/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounter.java#L86) or [KubernetesCheckpointIDCounter:167](https://github.com/apache/flink/blob/c6997c97c575d334679915c328792b8a3067cfb5/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/highavailability/KubernetesCheckpointIDCounter.java#L167). Can't we move this value in a single place like `CheckpointConfig.DEFAULT_INITIAL_CHECKPOINT_ID` and use it here as well like `CheckpointConfig.DEFAULT_INITIAL_CHECKPOINT_ID - 1`? This would make the intention of this
literal `0` more obvious.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
##########
@@ -104,6 +105,9 @@
private final MailboxExecutor mailboxExecutor;
private Counter numRecordsOutCounter;
+ // record endOfInput state to avoid duplicate prepareCommit on final notifyCheckpointComplete
+ // once FLIP-147 is fully operational all endOfInput processing needs to be removed
Review comment:
Is there a ticket for the removal which could be mentioned in the comment?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaCommitter.java
##########
@@ -64,19 +65,31 @@
.<FlinkKafkaInternalProducer<?, ?>>map(Recyclable::getObject)
.orElseGet(() -> getRecoveryProducer(committable));
producer.commitTransaction();
- recyclable.ifPresent(Recyclable::close);
- } catch (ProducerFencedException | InvalidTxnStateException e) {
- // That means we have committed this transaction before.
+ } catch (RetriableException e) {
LOG.warn(
- "Encountered error {} while recovering transaction {}. "
- + "Presumably this transaction has been already committed before",
- e,
- committable);
- recyclable.ifPresent(Recyclable::close);
- } catch (Throwable e) {
- LOG.warn("Cannot commit Kafka transaction, retrying.", e);
+ "Encountered retriable exception while committing {}.", transactionalId, e);
retryableCommittables.add(committable);
+ continue;
+ } catch (ProducerFencedException e) {
+ // initTransaction has been called on this transaction before
+ LOG.error(
+ "Transactions {} timed out or was overridden and data has been potentially lost.",
+ transactionalId,
+ e);
+ } catch (InvalidTxnStateException e) {
+ // This exception only occurs when aborting after a commit or vice versa.
+ // It does not appear on double commits or double aborts.
+ LOG.error(
+ "Transaction {} was previously canceled and data has been lost.",
+ committable,
+ e);
+ } catch (Exception e) {
Review comment:
Just to be sure we don't miss anything here: Why don't we handle `Throwable` here anymore?
##########
File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaWriter.java
##########
@@ -189,7 +191,7 @@ public void write(IN element, Context context) throws IOException {
final List<KafkaCommittable> committables =
Collections.singletonList(
KafkaCommittable.of(currentProducer, producerPool::add));
- LOG.info("Committing {} committables.", committables);
+ LOG.debug("Committing {} committables {}.", committables, flush);
Review comment:
This log message might be confusing. What does the `flush` flag say in the end? I would suggest making the log message more descriptive.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702844549
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
##########
@@ -104,6 +105,9 @@
private final MailboxExecutor mailboxExecutor;
private Counter numRecordsOutCounter;
+ // record endOfInput state to avoid duplicate prepareCommit on final notifyCheckpointComplete
+ // once FLIP-147 is fully operational all endOfInput processing needs to be removed
Review comment:
Not yet. I think the scope is not clear yet. It probably will be big refactoring of the sink.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddc241c3ce922b83edff8cd73fd3215ec5eda330 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562)
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddc241c3ce922b83edff8cd73fd3215ec5eda330 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
}, {
"hash" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632",
"triggerID" : "640fa7cee84eccde4c60157cfbc3d18e2d3a618f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4572e70e68e47877631d12907a767343845afa98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620)
* 640fa7cee84eccde4c60157cfbc3d18e2d3a618f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23632)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577)
* 9e722ba86feb67261c9a49a59d74ffea620b2aa5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] XComp commented on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
XComp commented on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913537258
Running the FLINK-23850 job on this PR's codebase results in the following error:
```
java.lang.RuntimeException: Failed to send data to Kafka
at org.apache.flink.connector.kafka.sink.KafkaWriter.lambda$null$0(KafkaWriter.java:131)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:338)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:324)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:789)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:741)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.kafka.shaded.org.apache.kafka.common.InvalidRecordException: This record has failed the validation on broker and hence be rejected.
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #17152:
URL: https://github.com/apache/flink/pull/17152#discussion_r702887199
##########
File path: flink-connectors/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
##########
@@ -123,6 +130,13 @@
private SharedReference<AtomicBoolean> failed;
private SharedReference<AtomicLong> lastCheckpointedRecord;
+ @Parameterized.Parameter public int run;
+
+ @Parameterized.Parameters
+ public static Set<Integer> getConfigurations() {
Review comment:
You should really start reviewing by commit :p This is just a tmp commit for testing.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4572e70e68e47877631d12907a767343845afa98 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23620)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #17152: [FLINK-24131][datastream/kafka] Reliably shutdown producer
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #17152:
URL: https://github.com/apache/flink/pull/17152#issuecomment-913235346
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23562",
"triggerID" : "ddc241c3ce922b83edff8cd73fd3215ec5eda330",
"triggerType" : "PUSH"
}, {
"hash" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577",
"triggerID" : "ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95",
"triggerType" : "PUSH"
}, {
"hash" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613",
"triggerID" : "9e722ba86feb67261c9a49a59d74ffea620b2aa5",
"triggerType" : "PUSH"
}, {
"hash" : "4572e70e68e47877631d12907a767343845afa98",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4572e70e68e47877631d12907a767343845afa98",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ddb35bd5f1aa7959febbd5a85954a8c3efe9bf95 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23577)
* 9e722ba86feb67261c9a49a59d74ffea620b2aa5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=23613)
* 4572e70e68e47877631d12907a767343845afa98 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org