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/05/31 17:18:15 UTC
[GitHub] [flink] rkhachatryan opened a new pull request #16035: [FLINK-21355][state/changelog] Log metadata
rkhachatryan opened a new pull request #16035:
URL: https://github.com/apache/flink/pull/16035
## What is the purpose of the change
Write metadata before state changes so that they can be applied on recovery.
## Verifying this change
TBD
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? no
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16035:
URL: https://github.com/apache/flink/pull/16035#discussion_r642614515
##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractStateChangeLogger.java
##########
@@ -100,6 +127,8 @@ protected void log(
return serializeRaw(
wrapper -> {
wrapper.writeByte(op.code);
+ // todo: wrapper.writeShort(stateId); and/or sort and write once (same for key, ns?)
+ wrapper.writeUTF(metaInfo.getName());
Review comment:
Writing state name for each change is sub-optimal, and there are two ways to avoid it:
1. write a mapping from state name to some id in the beginning, and write only id here
2. sort changes by state name and write it once
The first option is easier to implement but turned out more error-prone (on recovery).
The second option is also more efficient and can be extended to other dimensions as well (key, namespace).
However, I think it's easier to have this simpler version with recovery integrated (and maybe benchmarked) and then apply the optimization.
Ticket: FLINK-22944.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18462",
"triggerID" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"triggerType" : "PUSH"
}, {
"hash" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 79448d397d5f9c3d8774836cb5e1e5bca15e51d6 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18462)
* 4cfed99fe90130b1d873d7e2b2cc41d310ec5476 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan merged pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
rkhachatryan merged pull request #16035:
URL: https://github.com/apache/flink/pull/16035
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16035: [FLINK-21355][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16035:
URL: https://github.com/apache/flink/pull/16035#discussion_r642612519
##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractStateChangeLogger.java
##########
@@ -87,11 +98,27 @@ protected void log(
ThrowingConsumer<DataOutputViewStreamWrapper, IOException> dataWriter,
Ns ns)
throws IOException {
- // todo: log metadata (FLINK-22808)
+ logMetaIfNeeded();
stateChangelogWriter.append(
keyContext.getCurrentKeyGroupIndex(), serialize(op, ns, dataWriter));
}
+ private void logMetaIfNeeded() throws IOException {
+ if (metaDataWritten) {
+ return;
+ }
+ stateChangelogWriter.append(
+ COMMON_KEY_GROUP,
Review comment:
A negative keygroup number here is used to make writer write metadata in the very beginning of its current batch. That way, on recovery metadata is be read before the data changes, even if they were made to a smaller keygroup or come in because of rescaling.
I'd add a version of `append` without `keygroup` to `StateChangelogWriter`, but probably there are some better alternatives?
##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/AbstractStateChangeLogger.java
##########
@@ -100,6 +127,8 @@ protected void log(
return serializeRaw(
wrapper -> {
wrapper.writeByte(op.code);
+ // todo: wrapper.writeShort(stateId); and/or sort and write once (same for key, ns?)
+ wrapper.writeUTF(metaInfo.getName());
Review comment:
Writing state name for each change is sub-optimal, and there are two ways to avoid it:
1. write a mapping from state name to some id in the beginning, and write only id here
2. sort changes by state name and write it once
The first option is easier to implement but turned out more error-prone (on recovery).
The second option is also more efficient and can be extended to other dimensions as well (key, namespace).
However, I think it's easier to have this simpler version with recovery integrated (and maybe benchmarked) and then apply the optimization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18462",
"triggerID" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"triggerType" : "PUSH"
}, {
"hash" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464",
"triggerID" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cfed99fe90130b1d873d7e2b2cc41d310ec5476 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16035: [FLINK-21355][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 79448d397d5f9c3d8774836cb5e1e5bca15e51d6 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851606179
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 e53cf6c42f530c9a5dcb6b060384eaa32ae15bff (Sat Aug 28 12:18:38 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] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18462",
"triggerID" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"triggerType" : "PUSH"
}, {
"hash" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464",
"triggerID" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"triggerType" : "PUSH"
}, {
"hash" : "849b0b5dd3635b2a2dab3d38ccba8d849e365107",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18542",
"triggerID" : "849b0b5dd3635b2a2dab3d38ccba8d849e365107",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cfed99fe90130b1d873d7e2b2cc41d310ec5476 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464)
* 849b0b5dd3635b2a2dab3d38ccba8d849e365107 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18542)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18462",
"triggerID" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"triggerType" : "PUSH"
}, {
"hash" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464",
"triggerID" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"triggerType" : "PUSH"
}, {
"hash" : "849b0b5dd3635b2a2dab3d38ccba8d849e365107",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "849b0b5dd3635b2a2dab3d38ccba8d849e365107",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4cfed99fe90130b1d873d7e2b2cc41d310ec5476 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464)
* 849b0b5dd3635b2a2dab3d38ccba8d849e365107 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16035: [FLINK-21355][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851606179
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 79448d397d5f9c3d8774836cb5e1e5bca15e51d6 (Mon May 31 17:21:07 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
* **This pull request references an unassigned [Jira ticket](https://issues.apache.org/jira/browse/FLINK-21355).** According to the [code contribution guide](https://flink.apache.org/contributing/contribute-code.html), tickets need to be assigned before starting with the implementation work.
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #16035:
URL: https://github.com/apache/flink/pull/16035#discussion_r654257598
##########
File path: flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/StateChangeLoggerTestBase.java
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.changelog;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.changelog.SequenceNumber;
+import org.apache.flink.runtime.state.changelog.StateChangelogWriter;
+import org.apache.flink.runtime.state.heap.InternalKeyContextImpl;
+import org.apache.flink.state.changelog.AbstractStateChangeLogger.StateChangeOperation;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.state.changelog.AbstractStateChangeLogger.StateChangeOperation.METADATA;
+import static org.junit.Assert.assertEquals;
+
+abstract class StateChangeLoggerTestBase<Namespace> {
+ /** A basic test for appending the metadata on first state access. */
+ @Test
+ public void testMetadataOperationLogged() throws IOException {
+ TestingStateChangelogWriter writer = new TestingStateChangelogWriter();
+ InternalKeyContextImpl<String> keyContext =
+ new InternalKeyContextImpl<>(KeyGroupRange.of(1, 1000), 1000);
+ StateChangeLogger<String, Namespace> logger = getLogger(writer, keyContext);
+
+ List<Tuple2<Integer, StateChangeOperation>> expectedAppends = new ArrayList<>();
+ expectedAppends.add(
+ Tuple2.of(-1 /* see AbstractStateChangeLogger#COMMON_KEY_GROUP */, METADATA));
Review comment:
Yes, sorry (I changed it but then accidentially rolled back together with type changes for other comment).
##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java
##########
@@ -339,6 +343,13 @@ public void notifyCheckpointAborted(long checkpointId) throws Exception {
stateDesc.getClass(), this.getClass());
throw new FlinkRuntimeException(message);
}
+ RegisteredKeyValueStateBackendMetaInfo<N, S> meta =
Review comment:
> Or shall we change the code below?
> (StateSnapshotTransformer.StateSnapshotTransformFactory<SV>) snapshotTransformFactory);
Good idea!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16035:
URL: https://github.com/apache/flink/pull/16035#issuecomment-851611489
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18462",
"triggerID" : "79448d397d5f9c3d8774836cb5e1e5bca15e51d6",
"triggerType" : "PUSH"
}, {
"hash" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18464",
"triggerID" : "4cfed99fe90130b1d873d7e2b2cc41d310ec5476",
"triggerType" : "PUSH"
}, {
"hash" : "849b0b5dd3635b2a2dab3d38ccba8d849e365107",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18542",
"triggerID" : "849b0b5dd3635b2a2dab3d38ccba8d849e365107",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 849b0b5dd3635b2a2dab3d38ccba8d849e365107 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=18542)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] Myasuka commented on a change in pull request #16035: [FLINK-22808][state/changelog] Log metadata
Posted by GitBox <gi...@apache.org>.
Myasuka commented on a change in pull request #16035:
URL: https://github.com/apache/flink/pull/16035#discussion_r654110979
##########
File path: flink-state-backends/flink-statebackend-changelog/src/test/java/org/apache/flink/state/changelog/StateChangeLoggerTestBase.java
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.state.changelog;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.changelog.SequenceNumber;
+import org.apache.flink.runtime.state.changelog.StateChangelogWriter;
+import org.apache.flink.runtime.state.heap.InternalKeyContextImpl;
+import org.apache.flink.state.changelog.AbstractStateChangeLogger.StateChangeOperation;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.state.changelog.AbstractStateChangeLogger.StateChangeOperation.METADATA;
+import static org.junit.Assert.assertEquals;
+
+abstract class StateChangeLoggerTestBase<Namespace> {
+ /** A basic test for appending the metadata on first state access. */
+ @Test
+ public void testMetadataOperationLogged() throws IOException {
+ TestingStateChangelogWriter writer = new TestingStateChangelogWriter();
+ InternalKeyContextImpl<String> keyContext =
+ new InternalKeyContextImpl<>(KeyGroupRange.of(1, 1000), 1000);
+ StateChangeLogger<String, Namespace> logger = getLogger(writer, keyContext);
+
+ List<Tuple2<Integer, StateChangeOperation>> expectedAppends = new ArrayList<>();
+ expectedAppends.add(
+ Tuple2.of(-1 /* see AbstractStateChangeLogger#COMMON_KEY_GROUP */, METADATA));
Review comment:
It seems you did not change the magic number to `COMMON_KEY_GROUP` here in test code.
##########
File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/state/changelog/ChangelogKeyedStateBackend.java
##########
@@ -339,6 +343,13 @@ public void notifyCheckpointAborted(long checkpointId) throws Exception {
stateDesc.getClass(), this.getClass());
throw new FlinkRuntimeException(message);
}
+ RegisteredKeyValueStateBackendMetaInfo<N, S> meta =
Review comment:
To be honest, the raw use of `RegisteredKeyValueStateBackendMetaInfo` constructor makes me a bit confused.
I have tried to change `RegisteredKeyValueStateBackendMetaInfo<N, S>` to `RegisteredKeyValueStateBackendMetaInfo<N, S, SEV>` and it dose not need to change many files. Or shall we change the code below?
~~~ java
RegisteredKeyValueStateBackendMetaInfo<N, SV> meta =
new RegisteredKeyValueStateBackendMetaInfo(
stateDesc.getType(),
stateDesc.getName(),
namespaceSerializer,
stateDesc.getSerializer(),
(StateSnapshotTransformer.StateSnapshotTransformFactory<SV>) snapshotTransformFactory);
~~~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org