You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/03/30 00:46:28 UTC
[GitHub] [kafka] lihaosky commented on a change in pull request #11829: [2/N][emit final] add processor metadata to be committed with offset
lihaosky commented on a change in pull request #11829:
URL: https://github.com/apache/kafka/pull/11829#discussion_r838009581
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.kafka.streams.processor.internals;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * ProcessorMetadata to be access and populated by processor node. This will be committed along with
+ * offset. This metadata is mainly for windowed aggregation processor to store last emitted timestamp
+ * for now. Therefore, the supported metadata value type is only Long which is timestamp type.
+ */
+public class ProcessorMetadata {
+
+ private final Map<String, Long> metadata;
+
+ // Whether metadata should be committed. We only need to commit if metadata is updated via
+ // put() or set explicitly
+ private boolean needsCommit;
+
+ public static ProcessorMetadata emptyMetadata() {
Review comment:
I thought it's clear to create emptyMetadata like this, but I'm ok to use public constructor as well.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.kafka.streams.processor.internals;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * ProcessorMetadata to be access and populated by processor node. This will be committed along with
+ * offset. This metadata is mainly for windowed aggregation processor to store last emitted timestamp
+ * for now. Therefore, the supported metadata value type is only Long which is timestamp type.
+ */
+public class ProcessorMetadata {
+
+ private final Map<String, Long> metadata;
+
+ // Whether metadata should be committed. We only need to commit if metadata is updated via
+ // put() or set explicitly
+ private boolean needsCommit;
+
+ public static ProcessorMetadata emptyMetadata() {
+ return new ProcessorMetadata();
+ }
+
+ public static ProcessorMetadata with(final Map<String, Long> metadata) {
Review comment:
I can change to public constructor.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractProcessorContext.java
##########
@@ -254,4 +256,25 @@ public TaskType taskType() {
public String changelogFor(final String storeName) {
return stateManager().changelogFor(storeName);
}
+
+ @Override
+ public void addProcessorMetadataKeyValue(final String key, final long value) {
+ processorMetadata.addMetadata(key, value);
+ }
+
+ @Override
+ public Long getProcessorMetadataForKey(final String key) {
+ return processorMetadata.getMetadata(key);
+ }
+
+ @Override
+ public void setProcessorMetadata(final ProcessorMetadata metadata) {
Review comment:
This is used in restoration to set processor metadata as a whole after deserialization and merging. The per key setter is used by processor. Maybe we can get rid of `getProcessorMetadata` if it's not used.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -433,28 +447,18 @@ public void resume() {
case SUSPENDED:
final Map<TopicPartition, Long> partitionTimes = extractPartitionTimes();
- committableOffsets = new HashMap<>(consumedOffsets.size());
- for (final Map.Entry<TopicPartition, Long> entry : consumedOffsets.entrySet()) {
- final TopicPartition partition = entry.getKey();
- Long offset = partitionGroup.headRecordOffset(partition);
- if (offset == null) {
- try {
- offset = mainConsumer.position(partition);
- } catch (final TimeoutException error) {
- // the `consumer.position()` call should never block, because we know that we did process data
- // for the requested partition and thus the consumer should have a valid local position
- // that it can return immediately
-
- // hence, a `TimeoutException` indicates a bug and thus we rethrow it as fatal `IllegalStateException`
- throw new IllegalStateException(error);
- } catch (final KafkaException fatal) {
- throw new StreamsException(fatal);
- }
- }
+ // If there's processor metadata to be committed. We need to commit them to all
+ // input partitions
+ final Set<TopicPartition> partitionsNeedCommit = processorContext.getProcessorMetadata().needCommit() ?
+ inputPartitions() : consumedOffsets.keySet();
+ committableOffsets = new HashMap<>(partitionsNeedCommit.size());
+
+ for (final TopicPartition partition : partitionsNeedCommit) {
+ final Long offset = findOffset(partition);
final long partitionTime = partitionTimes.get(partition);
- committableOffsets.put(partition, new OffsetAndMetadata(offset, encodeTimestamp(partitionTime)));
+ committableOffsets.put(partition, new OffsetAndMetadata(offset,
+ TopicPartitionMetadata.with(partitionTime, processorContext.getProcessorMetadata()).encode()));
Review comment:
Existing windowed aggregation not using "emit final" can't upgrade to it since store format is different. Yes, it's a breaking changing if user wants to use it.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorMetadata.java
##########
@@ -0,0 +1,153 @@
+/*
+ * 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.kafka.streams.processor.internals;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * ProcessorMetadata to be access and populated by processor node. This will be committed along with
+ * offset. This metadata is mainly for windowed aggregation processor to store last emitted timestamp
+ * for now. Therefore, the supported metadata value type is only Long which is timestamp type.
+ */
+public class ProcessorMetadata {
+
+ private final Map<String, Long> metadata;
+
+ // Whether metadata should be committed. We only need to commit if metadata is updated via
+ // put() or set explicitly
+ private boolean needsCommit;
+
+ public static ProcessorMetadata emptyMetadata() {
+ return new ProcessorMetadata();
+ }
+
+ public static ProcessorMetadata with(final Map<String, Long> metadata) {
+ return new ProcessorMetadata(metadata);
+ }
+
+ private ProcessorMetadata() {
+ this(new HashMap<>());
+ }
+
+ private ProcessorMetadata(final Map<String, Long> metadata) {
+ this.metadata = metadata;
+ needsCommit = false;
+ }
+
+ public static ProcessorMetadata deserialize(final byte[] metaData) {
+ if (metaData == null || metaData.length == 0) {
+ return new ProcessorMetadata();
+ }
+
+ final ByteBuffer buffer = ByteBuffer.wrap(metaData);
+ final int entrySize = buffer.getInt();
+ final Map<String, Long> metadata = new HashMap<>(entrySize);
+ for (int i = 0; i < entrySize; i++) {
+ final int keySize = buffer.getInt();
+ final byte[] keyBytes = new byte[keySize];
+ buffer.get(keyBytes);
+ final Long value = buffer.getLong();
+ metadata.put(new String(keyBytes, StandardCharsets.UTF_8), value);
+ }
+ return new ProcessorMetadata(metadata);
+ }
+
+ public byte[] serialize() {
+ if (metadata.isEmpty()) {
+ return new byte[0];
+ }
+
+ int kvSize = 0;
+ for (final Map.Entry<String, Long> entry : metadata.entrySet()) {
+ kvSize += Integer.BYTES;
+ kvSize += entry.getKey().getBytes(StandardCharsets.UTF_8).length;
Review comment:
Good point!
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopicPartitionMetadata.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.kafka.streams.processor.internals;
+
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Metadata to be committed together with TopicPartition offset
+ */
+public class TopicPartitionMetadata {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TopicPartitionMetadata.class);
+
+ // visible for testing
+ static final byte LATEST_MAGIC_BYTE = 2;
+
+ private final long partitionTime;
+ private final ProcessorMetadata processorMetadata;
+
+ static TopicPartitionMetadata with(final long partitionTime, final ProcessorMetadata processorMetadata) {
Review comment:
I thought it's simpler to use factory for this simple pojo. But I can use public constructor if you think it's better.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -537,6 +542,7 @@ public void closeDirty() {
public void updateInputPartitions(final Set<TopicPartition> topicPartitions, final Map<String, List<String>> allTopologyNodesToSourceTopics) {
super.updateInputPartitions(topicPartitions, allTopologyNodesToSourceTopics);
partitionGroup.updatePartitions(topicPartitions, recordQueueCreator::createQueue);
+ processorContext.getProcessorMetadata().setNeedsCommit(true);
Review comment:
If new partitions are added, I think it's ok to not commit immediately since old partitions still have original metadata and they can be fetched/merged.
If there's partition deletion and system crashed before writing to new partitions, metadata will be gone. But I'm not sure if we need to handle that since user deleted old partitions containing metadata.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
##########
@@ -907,21 +913,33 @@ private void resetOffsetsIfNeededAndInitializeMetadata(final java.util.function.
}
}
- private void initializeTaskTime(final Map<TopicPartition, OffsetAndMetadata> offsetsAndMetadata) {
+ private void initializeTaskTimeAndProcessorMetadata(final Map<TopicPartition, OffsetAndMetadata> offsetsAndMetadata) {
+ ProcessorMetadata finalProcessMetadata = null;
Review comment:
Good point. We can do that.
##########
File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/TopicPartitionMetadata.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.kafka.streams.processor.internals;
+
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.Objects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Metadata to be committed together with TopicPartition offset
+ */
+public class TopicPartitionMetadata {
+
+ private static final Logger LOG = LoggerFactory.getLogger(TopicPartitionMetadata.class);
+
+ // visible for testing
+ static final byte LATEST_MAGIC_BYTE = 2;
+
+ private final long partitionTime;
+ private final ProcessorMetadata processorMetadata;
+
+ static TopicPartitionMetadata with(final long partitionTime, final ProcessorMetadata processorMetadata) {
+ return new TopicPartitionMetadata(partitionTime, processorMetadata);
+ }
+
+ private TopicPartitionMetadata(final long partitionTime, final ProcessorMetadata processorMetadata) {
+ Objects.requireNonNull(processorMetadata);
+ this.partitionTime = partitionTime;
+ this.processorMetadata = processorMetadata;
+ }
+
+ public long partitionTime() {
+ return partitionTime;
+ }
+
+ public ProcessorMetadata processorMetadata() {
+ return processorMetadata;
+ }
+
+ public String encode() {
+ final byte[] serializedMeta = processorMetadata.serialize();
+ // Format: MAGIC_BYTE(1) + PartitionTime(8) + processMeta
+ final ByteBuffer buffer = ByteBuffer.allocate(Byte.BYTES + Long.BYTES + serializedMeta.length);
+ buffer.put(LATEST_MAGIC_BYTE);
+ buffer.putLong(partitionTime);
+ buffer.put(serializedMeta);
+ return Base64.getEncoder().encodeToString(buffer.array());
Review comment:
The reason is after converting single partitionTime to String, the length is not deterministic and we need other meta to encode it. Also I don't want to use Base64 multiple times.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org