You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2023/05/04 12:55:02 UTC
[GitHub] [kafka] dajac opened a new pull request, #13675: KAFKA-14462; [14/N]; Add PartitionWriter
dajac opened a new pull request, #13675:
URL: https://github.com/apache/kafka/pull/13675
This patch introduces the `PartitionWriter` interface in the `group-coordinator` module. The `ReplicaManager` resides in the `core` module and it is thus not accessible from the `group-coordinator` one. The `PartitionWriterImpl` is basically an implementation of the interface residing in `core` which interfaces with the `ReplicaManager`.
One notable difference from the usual produce path is that the `PartitionWriter` returns the offset following the written records. This is then used by the coordinator runtime (coming later) to track when the request associated with the write can be completed.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212280128
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
Review Comment:
This return value is the offset? Can we add a comment?
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213474996
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordSerializer.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
+
+/**
+ * Serializer which serializes {{@link Record}} to bytes.
+ */
+public class RecordSerializer implements PartitionWriter.Serializer<Record> {
+ @Override
+ public byte[] serializeKey(Record record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
Review Comment:
I guess I was wondering why we have a unique class here for this. It makes it seem like we would want other implementations.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218596726
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.coordinator.group.runtime;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.List;
+
+/**
+ * A simple interface to write records to Partitions/Logs. It contains the minimum
+ * required for coordinators.
+ *
+ * @param <T> The record type.
+ */
+public interface PartitionWriter<T> {
+
+ /**
+ * Serializer to translate T to bytes.
+ *
+ * @param <T> The record type.
+ */
+ interface Serializer<T> {
+ /**
+ * Serializes the key of the record.
+ */
+ byte[] serializeKey(T record);
+
+ /**
+ * Serializes the value of the record.
+ */
+ byte[] serializeValue(T record);
+ }
+
+ /**
+ * Listener allowing to listen to high watermark changes. This is meant
+ * to be used in conjunction with {{@link PartitionWriter#append(TopicPartition, List)}}.
Review Comment:
It's 17/N right? I thought I saw some there.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218956441
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.coordinator.group.runtime;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.List;
+
+/**
+ * A simple interface to write records to Partitions/Logs. It contains the minimum
+ * required for coordinators.
+ *
+ * @param <T> The record type.
+ */
+public interface PartitionWriter<T> {
+
+ /**
+ * Serializer to translate T to bytes.
+ *
+ * @param <T> The record type.
+ */
+ interface Serializer<T> {
+ /**
+ * Serializes the key of the record.
+ */
+ byte[] serializeKey(T record);
+
+ /**
+ * Serializes the value of the record.
+ */
+ byte[] serializeValue(T record);
+ }
+
+ /**
+ * Listener allowing to listen to high watermark changes. This is meant
+ * to be used in conjunction with {{@link PartitionWriter#append(TopicPartition, List)}}.
Review Comment:
https://github.com/apache/kafka/pull/13795 is the one.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212340267
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordSerializer.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
+
+/**
+ * Serializer which serializes {{@link Record}} to bytes.
+ */
+public class RecordSerializer implements PartitionWriter.Serializer<Record> {
+ @Override
+ public byte[] serializeKey(Record record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
Review Comment:
Are there other ways to serialize? Just curious if we had something like this already.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218514447
##########
core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.ReplicaManager
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch}
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Properties}
+import scala.collection.Map
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class CoordinatorPartitionWriterTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new CoordinatorPartitionWriter(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdapter(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdapter(listener))
+
+ assertEquals(
+ new ListenerAdapter(listener),
+ new ListenerAdapter(listener)
+ )
+ assertEquals(
+ new ListenerAdapter(listener).hashCode(),
+ new ListenerAdapter(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new CoordinatorPartitionWriter(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit])
+
+ when(replicaManager.appendRecords(
+ ArgumentMatchers.eq(0L),
+ ArgumentMatchers.eq(1.toShort),
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ callbackCapture.capture(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenAnswer( _ => {
+ callbackCapture.getValue.apply(Map(
+ tp -> new PartitionResponse(
+ Errors.NONE,
Review Comment:
good catch. i missed adding this when a refactored the code. will add it back.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218957638
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
Review Comment:
yup. we will need to change this when we get to implementing the transaction offset commits. we will also need to change the Record to carry on the producer epoch/id, etc.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214113530
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) {
+ throw new IllegalStateException("records must be non-empty.")
+ }
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) {
+ recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ )
+ } else {
+ throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+ }
+
+ val appendResults = replicaManager.appendToLocalLog(
Review Comment:
Let me rename the class to `CoordinatorPartitionWriter` to make it explicit that this is used for/by coordinators. Does this work for you?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218512680
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
Review Comment:
acks=1 is to not avoid going to the purgatory. as we don’t go the purgatory, the timeout does not matter.
--
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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218594348
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.coordinator.group.runtime;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.List;
+
+/**
+ * A simple interface to write records to Partitions/Logs. It contains the minimum
+ * required for coordinators.
+ *
+ * @param <T> The record type.
+ */
+public interface PartitionWriter<T> {
+
+ /**
+ * Serializer to translate T to bytes.
+ *
+ * @param <T> The record type.
+ */
+ interface Serializer<T> {
+ /**
+ * Serializes the key of the record.
+ */
+ byte[] serializeKey(T record);
+
+ /**
+ * Serializes the value of the record.
+ */
+ byte[] serializeValue(T record);
+ }
+
+ /**
+ * Listener allowing to listen to high watermark changes. This is meant
+ * to be used in conjunction with {{@link PartitionWriter#append(TopicPartition, List)}}.
Review Comment:
which PR are you referring to: https://github.com/apache/kafka/pull/13795 or https://github.com/apache/kafka/pull/13812?
or do you mean 15/N which i don't see
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1217632767
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -666,7 +667,8 @@ class ReplicaManager(val config: KafkaConfig,
recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching,
transactionalId: String = null,
- transactionStatePartition: Option[Int] = None): Unit = {
+ transactionStatePartition: Option[Int] = None,
+ actionQueueAdd: (() => Unit) => Unit = actionQueue.add): Unit = {
Review Comment:
I used a different approach.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218516577
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordSerializerTest.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupMetadataKey;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupMetadataValue;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class RecordSerializerTest {
Review Comment:
it is not possible to create a record with a null key. this is why i haven’t added it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #13675:
URL: https://github.com/apache/kafka/pull/13675
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1217620168
##########
clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java:
##########
@@ -131,6 +131,7 @@ public Map<Errors, Integer> errorCounts() {
public static final class PartitionResponse {
public Errors error;
public long baseOffset;
+ public long lastOffset;
Review Comment:
Yeah, it is usually the case. Ss we don't use the auto-generated classes here (yet), we can do this.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1066,9 +1066,6 @@ class ReplicaManager(val config: KafkaConfig,
requiredAcks == -1 || requiredAcks == 1 || requiredAcks == 0
}
- /**
- * Append the messages to the local replica logs
Review Comment:
Removed it by mistake.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218515694
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
+ internalTopicsAllowed = true,
+ origin = AppendOrigin.COORDINATOR,
+ entriesPerPartition = Map(tp -> recordsBuilder.build()),
+ responseCallback = results => appendResults = results,
+ // We can directly complete the purgatories here because we don't hold
+ // any conflicting locks.
+ actionQueue = directActionQueue
+ )
+
+ val partitionResult = appendResults.getOrElse(tp,
+ throw new IllegalStateException(s"Append status $appendResults should have partition $tp."))
Review Comment:
never, hopefully. it is only here as a defensive measure in case of a bug.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213267178
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1019,13 +1029,14 @@ class ReplicaManager(val config: KafkaConfig,
}
/**
- * Append the messages to the local replica logs
+ * Append the messages to the local replica logs. ReplicaManager#appendRecords should usually be
+ * used instead of this method.
Review Comment:
I did not use it here because it put stuff to the action queue and I don't want/need this in my case. The callback is also annoying. Let me circle back on this to see if can do better. I wrote this a while ago and I admit that I went with the easiest solution back then.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212289663
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) {
+ throw new IllegalStateException("records must be non-empty.")
+ }
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) {
+ recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ )
+ } else {
+ throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+ }
+
+ val appendResults = replicaManager.appendToLocalLog(
Review Comment:
Are the partitions we are listening to just the __consumer_offsets partitions?
I think this class is pretty generic (which is fine), but this append origin being the coordinator makes it less generic.
I know we are in the coordinator/group path, but it might not be obvious without looking at the file path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #13675:
URL: https://github.com/apache/kafka/pull/13675#issuecomment-1574008136
Hey David -- whenever we update appendRecords we have to modify a ton of tests that mock the method. (Basically add a ton of any(), since the mocker doesn't get default args). There are 283 failing now 😅
See here for an example of a PR that updates them: https://github.com/apache/kafka/pull/13391/files
--
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
[GitHub] [kafka] dajac commented on pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #13675:
URL: https://github.com/apache/kafka/pull/13675#issuecomment-1574019307
Yeah, I have noticed all the failed tests. I will fix them.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214570874
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1019,13 +1029,14 @@ class ReplicaManager(val config: KafkaConfig,
}
/**
- * Append the messages to the local replica logs
+ * Append the messages to the local replica logs. ReplicaManager#appendRecords should usually be
+ * used instead of this method.
Review Comment:
lgtm :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218516113
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordSerializer.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
+
+/**
+ * Serializer which serializes {{@link Record}} to bytes.
+ */
+public class RecordSerializer implements PartitionWriter.Serializer<Record> {
+ @Override
+ public byte[] serializeKey(Record record) {
+ // Record does not accept a null key.
Review Comment:
we already have this in the record class so we don’t really need it here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218602539
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
Review Comment:
gotcha
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212323293
##########
core/src/test/scala/unit/kafka/coordinator/group/PartitionWriterImplTest.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.{LogAppendResult, ReplicaManager, RequestLocal}
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch, RecordConversionStats}
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LeaderHwChange, LogAppendInfo, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Optional, OptionalInt, Properties}
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class PartitionWriterImplTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdaptor(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdaptor(listener))
+
+ assertEquals(
+ new ListenerAdaptor(listener),
+ new ListenerAdaptor(listener)
+ )
+ assertEquals(
+ new ListenerAdaptor(listener).hashCode(),
+ new ListenerAdaptor(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ when(replicaManager.appendToLocalLog(
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ ArgumentMatchers.eq(1),
+ ArgumentMatchers.eq(RequestLocal.NoCaching)
+ )).thenReturn(Map(tp -> LogAppendResult(new LogAppendInfo(
+ Optional.empty(),
+ 10,
+ OptionalInt.empty(),
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordConversionStats.EMPTY,
+ CompressionType.NONE,
+ CompressionType.NONE,
+ -1,
+ -1,
+ false,
+ -1L,
+ Collections.emptyList(),
+ "",
+ LeaderHwChange.INCREASED
+ ))))
+
+ val records = List(
+ ("k0", "v0"),
+ ("k1", "v1"),
+ ("k2", "v2"),
+ )
+
+ assertEquals(11, partitionRecordWriter.append(
+ tp,
+ records.asJava
+ ))
+
+ verify(replicaManager).maybeCompletePurgatories(
+ tp,
+ LeaderHwChange.INCREASED
+ )
+
+ val batch = recordsCapture.getValue.getOrElse(tp,
+ throw new AssertionError(s"No records for $tp"))
+ assertEquals(1, batch.batches().asScala.toList.size)
+
+ val receivedRecords = batch.records.asScala.map { record =>
+ (
+ Charset.defaultCharset().decode(record.key).toString,
+ Charset.defaultCharset().decode(record.value).toString,
+ )
+ }.toList
+
+ assertEquals(records, receivedRecords)
+ }
+
+ @Test
+ def testWriteRecordTooLarge(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val maxBatchSize = 16384
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Map(TopicConfig.MAX_MESSAGE_BYTES_CONFIG -> maxBatchSize).asJava,
+ new Properties()
+ )))
+
+ val randomBytes = TestUtils.randomBytes(maxBatchSize + 1)
+ // We need more than one record here because the first record
+ // is always allowed by the MemoryRecordsBuilder.
Review Comment:
Is this a bug/quirk or is there a reason that the first record is always allowed?
--
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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218276954
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
Review Comment:
nit: "ListenerAdapter"
##########
core/src/test/scala/unit/kafka/coordinator/group/CoordinatorPartitionWriterTest.scala:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.ReplicaManager
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch}
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Properties}
+import scala.collection.Map
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class CoordinatorPartitionWriterTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new CoordinatorPartitionWriter(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdapter(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdapter(listener))
+
+ assertEquals(
+ new ListenerAdapter(listener),
+ new ListenerAdapter(listener)
+ )
+ assertEquals(
+ new ListenerAdapter(listener).hashCode(),
+ new ListenerAdapter(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new CoordinatorPartitionWriter(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ val callbackCapture: ArgumentCaptor[Map[TopicPartition, PartitionResponse] => Unit] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, PartitionResponse] => Unit])
+
+ when(replicaManager.appendRecords(
+ ArgumentMatchers.eq(0L),
+ ArgumentMatchers.eq(1.toShort),
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ callbackCapture.capture(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any(),
+ ArgumentMatchers.any()
+ )).thenAnswer( _ => {
+ callbackCapture.getValue.apply(Map(
+ tp -> new PartitionResponse(
+ Errors.NONE,
Review Comment:
we don't seem to do any validation of PartitionResponse error in CoordinatorPartitionWriter#appendRecords. what's the reason?
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
Review Comment:
will the offset manager also use this method to append records?
will the transaction coordinator use this to initiate txn offset commits?
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
+ internalTopicsAllowed = true,
+ origin = AppendOrigin.COORDINATOR,
+ entriesPerPartition = Map(tp -> recordsBuilder.build()),
+ responseCallback = results => appendResults = results,
+ // We can directly complete the purgatories here because we don't hold
+ // any conflicting locks.
+ actionQueue = directActionQueue
+ )
+
+ val partitionResult = appendResults.getOrElse(tp,
+ throw new IllegalStateException(s"Append status $appendResults should have partition $tp."))
Review Comment:
when do we expect to hit this exception?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.coordinator.group.runtime;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.List;
+
+/**
+ * A simple interface to write records to Partitions/Logs. It contains the minimum
+ * required for coordinators.
+ *
+ * @param <T> The record type.
+ */
+public interface PartitionWriter<T> {
+
+ /**
+ * Serializer to translate T to bytes.
+ *
+ * @param <T> The record type.
+ */
+ interface Serializer<T> {
+ /**
+ * Serializes the key of the record.
+ */
+ byte[] serializeKey(T record);
+
+ /**
+ * Serializes the value of the record.
+ */
+ byte[] serializeValue(T record);
+ }
+
+ /**
+ * Listener allowing to listen to high watermark changes. This is meant
+ * to be used in conjunction with {{@link PartitionWriter#append(TopicPartition, List)}}.
Review Comment:
my understanding is this gets registered to the Partition. once the hwm is updated, we trigger Listener#onHighWatermarkUpdated but we don't use in PartitionWriter#append and I don't see it used anywhere else.
Is there an example of what listener we actually register?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordSerializer.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
+
+/**
+ * Serializer which serializes {{@link Record}} to bytes.
+ */
+public class RecordSerializer implements PartitionWriter.Serializer<Record> {
+ @Override
+ public byte[] serializeKey(Record record) {
+ // Record does not accept a null key.
Review Comment:
should we require not null for record.key?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/RecordSerializerTest.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupMetadataKey;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupMetadataValue;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class RecordSerializerTest {
Review Comment:
should we add a test case for serializing a null key?
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
Review Comment:
to confirm, timeout=0 since we don't really care about delayed produce request and requiredAcks=1 to prevent `delayedProduceRequestRequired()` from returning true right?
this seems a bit confusing since we don't really care about these values
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214114184
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1019,13 +1029,14 @@ class ReplicaManager(val config: KafkaConfig,
}
/**
- * Append the messages to the local replica logs
+ * Append the messages to the local replica logs. ReplicaManager#appendRecords should usually be
+ * used instead of this method.
Review Comment:
I have refactored the implementation to use `appendRecords`. I think that it is better like this. Let me know what you think.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213273088
##########
core/src/test/scala/unit/kafka/coordinator/group/PartitionWriterImplTest.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.{LogAppendResult, ReplicaManager, RequestLocal}
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch, RecordConversionStats}
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LeaderHwChange, LogAppendInfo, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Optional, OptionalInt, Properties}
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class PartitionWriterImplTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdaptor(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdaptor(listener))
+
+ assertEquals(
+ new ListenerAdaptor(listener),
+ new ListenerAdaptor(listener)
+ )
+ assertEquals(
+ new ListenerAdaptor(listener).hashCode(),
+ new ListenerAdaptor(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ when(replicaManager.appendToLocalLog(
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ ArgumentMatchers.eq(1),
+ ArgumentMatchers.eq(RequestLocal.NoCaching)
+ )).thenReturn(Map(tp -> LogAppendResult(new LogAppendInfo(
+ Optional.empty(),
+ 10,
+ OptionalInt.empty(),
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordConversionStats.EMPTY,
+ CompressionType.NONE,
+ CompressionType.NONE,
+ -1,
+ -1,
+ false,
+ -1L,
+ Collections.emptyList(),
+ "",
+ LeaderHwChange.INCREASED
+ ))))
+
+ val records = List(
+ ("k0", "v0"),
+ ("k1", "v1"),
+ ("k2", "v2"),
+ )
+
+ assertEquals(11, partitionRecordWriter.append(
+ tp,
+ records.asJava
+ ))
+
+ verify(replicaManager).maybeCompletePurgatories(
+ tp,
+ LeaderHwChange.INCREASED
+ )
+
+ val batch = recordsCapture.getValue.getOrElse(tp,
+ throw new AssertionError(s"No records for $tp"))
+ assertEquals(1, batch.batches().asScala.toList.size)
+
+ val receivedRecords = batch.records.asScala.map { record =>
+ (
+ Charset.defaultCharset().decode(record.key).toString,
+ Charset.defaultCharset().decode(record.value).toString,
+ )
+ }.toList
+
+ assertEquals(records, receivedRecords)
+ }
+
+ @Test
+ def testWriteRecordTooLarge(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val maxBatchSize = 16384
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Map(TopicConfig.MAX_MESSAGE_BYTES_CONFIG -> maxBatchSize).asJava,
+ new Properties()
+ )))
+
+ val randomBytes = TestUtils.randomBytes(maxBatchSize + 1)
+ // We need more than one record here because the first record
+ // is always allowed by the MemoryRecordsBuilder.
Review Comment:
I actually don't know the reason. I only know that `MemoryRecordsBuilder` works this way...
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213268934
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) {
+ throw new IllegalStateException("records must be non-empty.")
+ }
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) {
+ recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ )
+ } else {
+ throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+ }
+
+ val appendResults = replicaManager.appendToLocalLog(
Review Comment:
Yeah, I agree. The alternative would be to pass it via the method but this would mean hardcoding it somewhere else. Is it better?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213267178
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1019,13 +1029,14 @@ class ReplicaManager(val config: KafkaConfig,
}
/**
- * Append the messages to the local replica logs
+ * Append the messages to the local replica logs. ReplicaManager#appendRecords should usually be
+ * used instead of this method.
Review Comment:
I did not use it here because it put stuff to the action queue and I don't want/need this in my case. Let me circle back on this to see if can do better. I wrote this a while ago and I admit that I went with the easiest solution back then.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214109717
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordSerializer.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
+
+/**
+ * Serializer which serializes {{@link Record}} to bytes.
+ */
+public class RecordSerializer implements PartitionWriter.Serializer<Record> {
+ @Override
+ public byte[] serializeKey(Record record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
Review Comment:
Yeah. I made it generic in the case we would want to reuse it for another coordinator in the future. This one is tight to the Record class.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214569080
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -131,26 +132,22 @@ class PartitionWriterImpl[T](
s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
}
- val appendResults = replicaManager.appendToLocalLog(
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
internalTopicsAllowed = true,
origin = AppendOrigin.COORDINATOR,
entriesPerPartition = Map(tp -> recordsBuilder.build()),
- requiredAcks = 1,
- requestLocal = RequestLocal.NoCaching
+ responseCallback = results => appendResults = results,
+ actionQueueAdd = item => item() // Immediately complete the action queue item.
)
val partitionResult = appendResults.getOrElse(tp,
- throw new IllegalStateException("Append status %s should only have one partition %s"
- .format(appendResults, tp)))
-
- // Complete delayed operations.
- replicaManager.maybeCompletePurgatories(
- tp,
- partitionResult.info.leaderHwChange
- )
+ throw new IllegalStateException(s"Append status $appendResults should only have one partition $tp"))
Review Comment:
nit: this error message confused me because I thought that it was saying there was more than one tp and that's why it failed. I think maybe the message should say something like s"Append status $appendResults should have partition $tp" since the fact that it doesn't would be the reason to throw the error.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212323728
##########
core/src/test/scala/unit/kafka/coordinator/group/PartitionWriterImplTest.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.{LogAppendResult, ReplicaManager, RequestLocal}
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch, RecordConversionStats}
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LeaderHwChange, LogAppendInfo, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Optional, OptionalInt, Properties}
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class PartitionWriterImplTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdaptor(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdaptor(listener))
+
+ assertEquals(
+ new ListenerAdaptor(listener),
+ new ListenerAdaptor(listener)
+ )
+ assertEquals(
+ new ListenerAdaptor(listener).hashCode(),
+ new ListenerAdaptor(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ when(replicaManager.appendToLocalLog(
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ ArgumentMatchers.eq(1),
+ ArgumentMatchers.eq(RequestLocal.NoCaching)
+ )).thenReturn(Map(tp -> LogAppendResult(new LogAppendInfo(
+ Optional.empty(),
+ 10,
+ OptionalInt.empty(),
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordConversionStats.EMPTY,
+ CompressionType.NONE,
+ CompressionType.NONE,
+ -1,
+ -1,
+ false,
+ -1L,
+ Collections.emptyList(),
+ "",
+ LeaderHwChange.INCREASED
+ ))))
+
+ val records = List(
+ ("k0", "v0"),
+ ("k1", "v1"),
+ ("k2", "v2"),
+ )
+
+ assertEquals(11, partitionRecordWriter.append(
+ tp,
+ records.asJava
+ ))
+
+ verify(replicaManager).maybeCompletePurgatories(
+ tp,
+ LeaderHwChange.INCREASED
+ )
+
+ val batch = recordsCapture.getValue.getOrElse(tp,
+ throw new AssertionError(s"No records for $tp"))
+ assertEquals(1, batch.batches().asScala.toList.size)
+
+ val receivedRecords = batch.records.asScala.map { record =>
+ (
+ Charset.defaultCharset().decode(record.key).toString,
+ Charset.defaultCharset().decode(record.value).toString,
+ )
+ }.toList
+
+ assertEquals(records, receivedRecords)
+ }
+
+ @Test
+ def testWriteRecordTooLarge(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val maxBatchSize = 16384
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Map(TopicConfig.MAX_MESSAGE_BYTES_CONFIG -> maxBatchSize).asJava,
+ new Properties()
+ )))
+
+ val randomBytes = TestUtils.randomBytes(maxBatchSize + 1)
+ // We need more than one record here because the first record
+ // is always allowed by the MemoryRecordsBuilder.
+ val records = List(
+ ("k0", new String(randomBytes)),
+ ("k1", new String(randomBytes)),
+ )
+
+ assertThrows(classOf[RecordTooLargeException],
+ () => partitionRecordWriter.append(tp, records.asJava))
+ }
+
+ @Test
+ def testWriteEmptyRecordList(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ assertThrows(classOf[IllegalStateException],
+ () => partitionRecordWriter.append(tp, List.empty.asJava))
+ }
+
+ @Test
+ def testInexistentPartition(): Unit = {
Review Comment:
nit: Nonexistent
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213473971
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1019,13 +1029,14 @@ class ReplicaManager(val config: KafkaConfig,
}
/**
- * Append the messages to the local replica logs
+ * Append the messages to the local replica logs. ReplicaManager#appendRecords should usually be
+ * used instead of this method.
Review Comment:
Those reasons make sense. The append path is pretty complex, so an overall refactor could do some good. But doesn't need to be a blocker here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214829515
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.ReplicaManager
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
Review Comment:
do we not need anything in the header because these are only internal topics?
what metadata do other record types contain in the header, client info?
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.ReplicaManager
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
+ internalTopicsAllowed = true,
+ origin = AppendOrigin.COORDINATOR,
+ entriesPerPartition = Map(tp -> recordsBuilder.build()),
+ responseCallback = results => appendResults = results,
+ actionQueueAdd = item => item() // Immediately complete the action queue item.
Review Comment:
my understanding is that the actionQueue and ActionQueue#tryCompleteActions() were created to resolve issues related to concurrency.
is it safe here because the threads from MultiThreadedEventProcessor do not hold locks and can we add a comment on why?
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1066,9 +1066,6 @@ class ReplicaManager(val config: KafkaConfig,
requiredAcks == -1 || requiredAcks == 1 || requiredAcks == 0
}
- /**
- * Append the messages to the local replica logs
Review Comment:
what's the reason for removing this?
##########
clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java:
##########
@@ -131,6 +131,7 @@ public Map<Errors, Integer> errorCounts() {
public static final class PartitionResponse {
public Errors error;
public long baseOffset;
+ public long lastOffset;
Review Comment:
i always assumed that there was a 1-1 mapping between the json (ProduceResponse.json) and the internal representation. i'm guessing we don't have to change the actual protocol since we only use this value internally?
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.ReplicaManager
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
Review Comment:
nit: should we stick with Adapter? like we do for GroupCoordinatorAdapter
also, can we have a comment on what this does or a link to PartitionListener
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -666,7 +667,8 @@ class ReplicaManager(val config: KafkaConfig,
recordConversionStatsCallback: Map[TopicPartition, RecordConversionStats] => Unit = _ => (),
requestLocal: RequestLocal = RequestLocal.NoCaching,
transactionalId: String = null,
- transactionStatePartition: Option[Int] = None): Unit = {
+ transactionStatePartition: Option[Int] = None,
+ actionQueueAdd: (() => Unit) => Unit = actionQueue.add): Unit = {
Review Comment:
how's actionQueueOperation? we don't add() in CoordinatorPartitionWriter right
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213271843
##########
core/src/test/scala/unit/kafka/coordinator/group/PartitionWriterImplTest.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.{LogAppendResult, ReplicaManager, RequestLocal}
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch, RecordConversionStats}
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LeaderHwChange, LogAppendInfo, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Optional, OptionalInt, Properties}
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class PartitionWriterImplTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdaptor(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdaptor(listener))
+
+ assertEquals(
+ new ListenerAdaptor(listener),
+ new ListenerAdaptor(listener)
+ )
+ assertEquals(
+ new ListenerAdaptor(listener).hashCode(),
+ new ListenerAdaptor(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ when(replicaManager.appendToLocalLog(
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ ArgumentMatchers.eq(1),
+ ArgumentMatchers.eq(RequestLocal.NoCaching)
+ )).thenReturn(Map(tp -> LogAppendResult(new LogAppendInfo(
+ Optional.empty(),
+ 10,
+ OptionalInt.empty(),
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordConversionStats.EMPTY,
+ CompressionType.NONE,
+ CompressionType.NONE,
+ -1,
+ -1,
+ false,
+ -1L,
+ Collections.emptyList(),
+ "",
+ LeaderHwChange.INCREASED
+ ))))
+
+ val records = List(
+ ("k0", "v0"),
+ ("k1", "v1"),
+ ("k2", "v2"),
+ )
+
+ assertEquals(11, partitionRecordWriter.append(
+ tp,
+ records.asJava
+ ))
+
+ verify(replicaManager).maybeCompletePurgatories(
+ tp,
+ LeaderHwChange.INCREASED
+ )
+
+ val batch = recordsCapture.getValue.getOrElse(tp,
+ throw new AssertionError(s"No records for $tp"))
+ assertEquals(1, batch.batches().asScala.toList.size)
+
+ val receivedRecords = batch.records.asScala.map { record =>
+ (
+ Charset.defaultCharset().decode(record.key).toString,
+ Charset.defaultCharset().decode(record.value).toString,
+ )
+ }.toList
+
+ assertEquals(records, receivedRecords)
+ }
+
+ @Test
+ def testWriteRecordTooLarge(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val maxBatchSize = 16384
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Map(TopicConfig.MAX_MESSAGE_BYTES_CONFIG -> maxBatchSize).asJava,
+ new Properties()
+ )))
+
+ val randomBytes = TestUtils.randomBytes(maxBatchSize + 1)
+ // We need more than one record here because the first record
+ // is always allowed by the MemoryRecordsBuilder.
+ val records = List(
+ ("k0", new String(randomBytes)),
+ ("k1", new String(randomBytes)),
+ )
+
+ assertThrows(classOf[RecordTooLargeException],
+ () => partitionRecordWriter.append(tp, records.asJava))
+ }
+
+ @Test
+ def testWriteEmptyRecordList(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ assertThrows(classOf[IllegalStateException],
+ () => partitionRecordWriter.append(tp, List.empty.asJava))
+ }
+
+ @Test
+ def testInexistentPartition(): Unit = {
Review Comment:
Right but the config does not exist because the partition does not exist either.
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218576567
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
Review Comment:
it is not a workaround. this is exactly what acks=1 is for, even for regular produce request. adding default values would be risky here. it is much better to let the caller be explicit about what he needs/wants.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1214564305
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) {
+ throw new IllegalStateException("records must be non-empty.")
+ }
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) {
+ recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ )
+ } else {
+ throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+ }
+
+ val appendResults = replicaManager.appendToLocalLog(
Review Comment:
Sounds good. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1217619263
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,156 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.ReplicaManager
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
Review Comment:
We don't use headers at all in the group coordinator. Headers can be used to stored anything. For regular topics, they are defined by the end users on the Producer.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212324258
##########
core/src/test/scala/unit/kafka/coordinator/group/PartitionWriterImplTest.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.server.{LogAppendResult, ReplicaManager, RequestLocal}
+import kafka.utils.TestUtils
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.config.TopicConfig
+import org.apache.kafka.common.errors.{NotLeaderOrFollowerException, RecordTooLargeException}
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, RecordBatch, RecordConversionStats}
+import org.apache.kafka.common.utils.{MockTime, Time}
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.{AppendOrigin, LeaderHwChange, LogAppendInfo, LogConfig}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows}
+import org.junit.jupiter.api.Test
+import org.mockito.{ArgumentCaptor, ArgumentMatchers}
+import org.mockito.Mockito.{mock, verify, when}
+
+import java.nio.charset.Charset
+import java.util.{Collections, Optional, OptionalInt, Properties}
+import scala.jdk.CollectionConverters._
+
+class StringKeyValueSerializer extends PartitionWriter.Serializer[(String, String)] {
+ override def serializeKey(record: (String, String)): Array[Byte] = {
+ record._1.getBytes(Charset.defaultCharset())
+ }
+
+ override def serializeValue(record: (String, String)): Array[Byte] = {
+ record._2.getBytes(Charset.defaultCharset())
+ }
+}
+
+class PartitionWriterImplTest {
+ @Test
+ def testRegisterDeregisterListener(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val listener = new PartitionWriter.Listener {
+ override def onHighWatermarkUpdated(tp: TopicPartition, offset: Long): Unit = {}
+ }
+
+ partitionRecordWriter.registerListener(tp, listener)
+ verify(replicaManager).maybeAddListener(tp, new ListenerAdaptor(listener))
+
+ partitionRecordWriter.deregisterListener(tp, listener)
+ verify(replicaManager).removeListener(tp, new ListenerAdaptor(listener))
+
+ assertEquals(
+ new ListenerAdaptor(listener),
+ new ListenerAdaptor(listener)
+ )
+ assertEquals(
+ new ListenerAdaptor(listener).hashCode(),
+ new ListenerAdaptor(listener).hashCode()
+ )
+ }
+
+ @Test
+ def testWriteRecords(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val time = new MockTime()
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ time
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ val recordsCapture: ArgumentCaptor[Map[TopicPartition, MemoryRecords]] =
+ ArgumentCaptor.forClass(classOf[Map[TopicPartition, MemoryRecords]])
+ when(replicaManager.appendToLocalLog(
+ ArgumentMatchers.eq(true),
+ ArgumentMatchers.eq(AppendOrigin.COORDINATOR),
+ recordsCapture.capture(),
+ ArgumentMatchers.eq(1),
+ ArgumentMatchers.eq(RequestLocal.NoCaching)
+ )).thenReturn(Map(tp -> LogAppendResult(new LogAppendInfo(
+ Optional.empty(),
+ 10,
+ OptionalInt.empty(),
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordBatch.NO_TIMESTAMP,
+ -1L,
+ RecordConversionStats.EMPTY,
+ CompressionType.NONE,
+ CompressionType.NONE,
+ -1,
+ -1,
+ false,
+ -1L,
+ Collections.emptyList(),
+ "",
+ LeaderHwChange.INCREASED
+ ))))
+
+ val records = List(
+ ("k0", "v0"),
+ ("k1", "v1"),
+ ("k2", "v2"),
+ )
+
+ assertEquals(11, partitionRecordWriter.append(
+ tp,
+ records.asJava
+ ))
+
+ verify(replicaManager).maybeCompletePurgatories(
+ tp,
+ LeaderHwChange.INCREASED
+ )
+
+ val batch = recordsCapture.getValue.getOrElse(tp,
+ throw new AssertionError(s"No records for $tp"))
+ assertEquals(1, batch.batches().asScala.toList.size)
+
+ val receivedRecords = batch.records.asScala.map { record =>
+ (
+ Charset.defaultCharset().decode(record.key).toString,
+ Charset.defaultCharset().decode(record.value).toString,
+ )
+ }.toList
+
+ assertEquals(records, receivedRecords)
+ }
+
+ @Test
+ def testWriteRecordTooLarge(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ val maxBatchSize = 16384
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Map(TopicConfig.MAX_MESSAGE_BYTES_CONFIG -> maxBatchSize).asJava,
+ new Properties()
+ )))
+
+ val randomBytes = TestUtils.randomBytes(maxBatchSize + 1)
+ // We need more than one record here because the first record
+ // is always allowed by the MemoryRecordsBuilder.
+ val records = List(
+ ("k0", new String(randomBytes)),
+ ("k1", new String(randomBytes)),
+ )
+
+ assertThrows(classOf[RecordTooLargeException],
+ () => partitionRecordWriter.append(tp, records.asJava))
+ }
+
+ @Test
+ def testWriteEmptyRecordList(): Unit = {
+ val tp = new TopicPartition("foo", 0)
+ val replicaManager = mock(classOf[ReplicaManager])
+ val partitionRecordWriter = new PartitionWriterImpl(
+ replicaManager,
+ new StringKeyValueSerializer(),
+ CompressionType.NONE,
+ Time.SYSTEM
+ )
+
+ when(replicaManager.getLogConfig(tp)).thenReturn(Some(LogConfig.fromProps(
+ Collections.emptyMap(),
+ new Properties()
+ )))
+
+ assertThrows(classOf[IllegalStateException],
+ () => partitionRecordWriter.append(tp, List.empty.asJava))
+ }
+
+ @Test
+ def testInexistentPartition(): Unit = {
Review Comment:
Also this is testing the log config doesn't exist right? Is this only the case if the partition is also nonexistent?
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212336472
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
Review Comment:
I guess this is shared in the interface, but maybe wouldn't hurt to say here as well.
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213469954
##########
core/src/main/scala/kafka/coordinator/group/PartitionWriterImpl.scala:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ReplicaManager, RequestLocal}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+private[group] class ListenerAdaptor(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdaptor => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class PartitionWriterImpl[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdaptor(listener))
+ }
+
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) {
+ throw new IllegalStateException("records must be non-empty.")
+ }
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) {
+ recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ )
+ } else {
+ throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+ }
+
+ val appendResults = replicaManager.appendToLocalLog(
Review Comment:
I'm not sure if it is maybe better to include a comment or include it somehow in the name. Hard coding is fine as long as it isn't a surprise 😅
--
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
[GitHub] [kafka] jolshan commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1212290566
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -1019,13 +1029,14 @@ class ReplicaManager(val config: KafkaConfig,
}
/**
- * Append the messages to the local replica logs
+ * Append the messages to the local replica logs. ReplicaManager#appendRecords should usually be
+ * used instead of this method.
Review Comment:
Is there a strong reason why appendRecords is preferred, but in this case we don't use it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1213274631
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/RecordSerializer.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * 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.coordinator.group;
+
+import org.apache.kafka.common.protocol.MessageUtil;
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter;
+
+/**
+ * Serializer which serializes {{@link Record}} to bytes.
+ */
+public class RecordSerializer implements PartitionWriter.Serializer<Record> {
+ @Override
+ public byte[] serializeKey(Record record) {
+ // Record does not accept a null key.
+ return MessageUtil.toVersionPrefixedBytes(
Review Comment:
We use the very same code `MessageUtil.toVersionPrefixedBytes` in scala.
--
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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218572827
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
+ tp: TopicPartition,
+ records: util.List[T]
+ ): Long = {
+ if (records.isEmpty) throw new IllegalStateException("records must be non-empty.")
+
+ replicaManager.getLogConfig(tp) match {
+ case Some(logConfig) =>
+ val magic = logConfig.recordVersion.value
+ val maxBatchSize = logConfig.maxMessageSize
+ val currentTimeMs = time.milliseconds()
+
+ val recordsBuilder = MemoryRecords.builder(
+ ByteBuffer.allocate(math.min(16384, maxBatchSize)),
+ magic,
+ compressionType,
+ TimestampType.CREATE_TIME,
+ 0L,
+ maxBatchSize
+ )
+
+ records.forEach { record =>
+ val keyBytes = serializer.serializeKey(record)
+ val valueBytes = serializer.serializeValue(record)
+
+ if (recordsBuilder.hasRoomFor(currentTimeMs, keyBytes, valueBytes, EMPTY_HEADERS)) recordsBuilder.append(
+ currentTimeMs,
+ keyBytes,
+ valueBytes,
+ EMPTY_HEADERS
+ ) else throw new RecordTooLargeException(s"Message batch size is ${recordsBuilder.estimatedSizeInBytes()} bytes " +
+ s"in append to partition $tp which exceeds the maximum configured size of $maxBatchSize.")
+ }
+
+ var appendResults: Map[TopicPartition, PartitionResponse] = Map.empty
+ replicaManager.appendRecords(
+ timeout = 0L,
+ requiredAcks = 1,
Review Comment:
right, which means it's kind of a workaround and is confusing because the acks setting is unrelated to how we're using appendRecords as well as the timeout.
i'm wondering if we can set the default values in appendRecords for acks and timeout
--
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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218578994
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
Review Comment:
the memory records builder used below in L126 assumes RecordBatch.NO_PRODUCER_EPOCH - do we need this once we introduce txn offset commits?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218511238
##########
core/src/main/scala/kafka/coordinator/group/CoordinatorPartitionWriter.scala:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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 kafka.coordinator.group
+
+import kafka.cluster.PartitionListener
+import kafka.server.{ActionQueue, ReplicaManager}
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.RecordTooLargeException
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.record.{CompressionType, MemoryRecords, TimestampType}
+import org.apache.kafka.common.record.Record.EMPTY_HEADERS
+import org.apache.kafka.common.requests.ProduceResponse.PartitionResponse
+import org.apache.kafka.common.utils.Time
+import org.apache.kafka.coordinator.group.runtime.PartitionWriter
+import org.apache.kafka.storage.internals.log.AppendOrigin
+
+import java.nio.ByteBuffer
+import java.util
+import scala.collection.Map
+
+/**
+ * ListenerAdapter adapts the PartitionListener interface to the
+ * PartitionWriter.Listener interface.
+ */
+private[group] class ListenerAdapter(
+ val listener: PartitionWriter.Listener
+) extends PartitionListener {
+ override def onHighWatermarkUpdated(
+ tp: TopicPartition,
+ offset: Long
+ ): Unit = {
+ listener.onHighWatermarkUpdated(tp, offset)
+ }
+
+ override def equals(that: Any): Boolean = that match {
+ case other: ListenerAdapter => listener.equals(other.listener)
+ case _ => false
+ }
+
+ override def hashCode(): Int = {
+ listener.hashCode()
+ }
+
+ override def toString: String = {
+ s"ListenerAdaptor(listener=$listener)"
+ }
+}
+
+class CoordinatorPartitionWriter[T](
+ replicaManager: ReplicaManager,
+ serializer: PartitionWriter.Serializer[T],
+ compressionType: CompressionType,
+ time: Time
+) extends PartitionWriter[T] {
+ // We use an action queue which directly executes actions. This is possible
+ // here because we don't hold any conflicting locks.
+ private val directActionQueue = new ActionQueue {
+ override def add(action: () => Unit): Unit = {
+ action()
+ }
+
+ override def tryCompleteActions(): Unit = {}
+ }
+
+ /**
+ * Register a PartitionWriter.Listener.
+ *
+ * @param tp The partition to register the listener to.
+ * @param listener The listener.
+ */
+ override def registerListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.maybeAddListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Deregister a PartitionWriter.Listener.
+ *
+ * @param tp The partition to deregister the listener from.
+ * @param listener The listener.
+ */
+ override def deregisterListener(
+ tp: TopicPartition,
+ listener: PartitionWriter.Listener
+ ): Unit = {
+ replicaManager.removeListener(tp, new ListenerAdapter(listener))
+ }
+
+ /**
+ * Write records to the partitions. Records are written in one batch so
+ * atomicity is guaranteed.
+ *
+ * @param tp The partition to write records to.
+ * @param records The list of records. The records are written in a single batch.
+ * @return The log end offset right after the written records.
+ * @throws KafkaException Any KafkaException caught during the write operation.
+ */
+ override def append(
Review Comment:
All the writes done be the group coordinator will go through this. Why are you asking?
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1218517345
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.coordinator.group.runtime;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.List;
+
+/**
+ * A simple interface to write records to Partitions/Logs. It contains the minimum
+ * required for coordinators.
+ *
+ * @param <T> The record type.
+ */
+public interface PartitionWriter<T> {
+
+ /**
+ * Serializer to translate T to bytes.
+ *
+ * @param <T> The record type.
+ */
+ interface Serializer<T> {
+ /**
+ * Serializes the key of the record.
+ */
+ byte[] serializeKey(T record);
+
+ /**
+ * Serializes the value of the record.
+ */
+ byte[] serializeValue(T record);
+ }
+
+ /**
+ * Listener allowing to listen to high watermark changes. This is meant
+ * to be used in conjunction with {{@link PartitionWriter#append(TopicPartition, List)}}.
Review Comment:
you can check out the next pr to see how this is used.
--
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
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #13675:
URL: https://github.com/apache/kafka/pull/13675#discussion_r1219669199
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/PartitionWriter.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * 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.coordinator.group.runtime;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.TopicPartition;
+
+import java.util.List;
+
+/**
+ * A simple interface to write records to Partitions/Logs. It contains the minimum
+ * required for coordinators.
+ *
+ * @param <T> The record type.
+ */
+public interface PartitionWriter<T> {
+
+ /**
+ * Serializer to translate T to bytes.
+ *
+ * @param <T> The record type.
+ */
+ interface Serializer<T> {
+ /**
+ * Serializes the key of the record.
+ */
+ byte[] serializeKey(T record);
+
+ /**
+ * Serializes the value of the record.
+ */
+ byte[] serializeValue(T record);
+ }
+
+ /**
+ * Listener allowing to listen to high watermark changes. This is meant
+ * to be used in conjunction with {{@link PartitionWriter#append(TopicPartition, List)}}.
Review Comment:
ah found HighWatermarkListener. thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #13675: KAFKA-14462; [14/N] Add PartitionWriter
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #13675:
URL: https://github.com/apache/kafka/pull/13675#issuecomment-1577150595
These connect tests are a little concerning. They seem unrelated but maybe we should file a jira.
--
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