pnowojski commented on a change in pull request #14838: URL: https://github.com/apache/flink/pull/14838#discussion_r574306144
########## File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/changelog/StateChangelogWriter.java ########## @@ -0,0 +1,77 @@ +/* + * 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.changelog; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.state.StateChangelogHandle; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +/** Allows to write data to the log. Scoped to a single writer (e.g. state backend). */ +@Internal +public interface StateChangelogWriter<Handle extends StateChangelogHandle<?>> + extends AutoCloseable { + + /** + * Get {@link SequenceNumber} of the last element added by {@link #append(int, byte[]) append}. + */ + SequenceNumber lastAppendedSqn(); + + /** Appends the provided data to this log. No persistency guarantees. */ + void append(int keyGroup, byte[] value); + + /** + * Durably persist previously {@link #append(int, byte[]) appended} data. After this call, one + * of {@link #confirm(SequenceNumber, SequenceNumber) confirm}, {@link #reset(SequenceNumber, + * SequenceNumber) reset}, or {@link #truncate(SequenceNumber) truncate} eventually must be + * called for the corresponding change set. with reset/truncate/confirm methods? + * + * @param from inclusive + */ + CompletableFuture<Handle> persist(SequenceNumber from) throws IOException; Review comment: Why is this "persist from"? From where until what? Shouldn't this be "persist until" or "persist up to"? If not, then please explain this in the javadoc? ########## File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/changelog/StateChangelogWriter.java ########## @@ -0,0 +1,77 @@ +/* + * 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.changelog; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.state.StateChangelogHandle; + +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +/** Allows to write data to the log. Scoped to a single writer (e.g. state backend). */ +@Internal +public interface StateChangelogWriter<Handle extends StateChangelogHandle<?>> + extends AutoCloseable { + + /** + * Get {@link SequenceNumber} of the last element added by {@link #append(int, byte[]) append}. + */ + SequenceNumber lastAppendedSqn(); Review comment: `lastAppendedSqn` -> lastAppendedSequenceNumber` ########## File path: flink-state-backends/pom.xml ########## @@ -39,5 +39,6 @@ under the License. <modules> <module>flink-statebackend-rocksdb</module> <module>flink-statebackend-heap-spillable</module> + <module>flink-statebackend-changelog</module> Review comment: I think the interfaces definitions should be in the `flink-runtime`, probably along the in memory implementation. We know there are other usecases where we might want to use it, furthermore there will be different implementations of the changelog. Maybe file based implementation could be placed anywhere for the time being, but I would need to think about this structure a bit more. ########## File path: flink-state-backends/flink-statebackend-changelog/src/main/java/org/apache/flink/changelog/StateChangelogClient.java ########## @@ -0,0 +1,38 @@ +/* + * 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.changelog; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.jobgraph.OperatorID; +import org.apache.flink.runtime.state.KeyGroupRange; +import org.apache.flink.runtime.state.StateChangelogHandle; + +/** + * {@link StateChangelogWriter} factory. Scoped to a single entity (e.g. a SubTask or + * OperatorCoordinator). + */ +@Internal +public interface StateChangelogClient<Handle extends StateChangelogHandle<?>> Review comment: Why is factory class named client? I had no idea what `StateChangelogClient` is doing looking at the name (I was more expecting it to be an internal class used by the changelog to perform the writes), and it's a bit funny that javadoc makes it easy to understand in the first sentence by calling it a `factory` :) ---------------------------------------------------------------- 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: [email protected]
