Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions build.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -858,6 +858,7 @@ project(':core') {
implementation project(':group-coordinator')
implementation project(':metadata')
implementation project(':storage:api')
implementation project(':tools:tools-api')

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Gradle can't distinguish the subproject name (gradle/gradle#847), so I add the prefix tools to avoid conflict to storage:api

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why does core depend on tools-api?

implementation project(':raft')
implementation project(':storage')

Expand Down Expand Up @@ -1106,6 +1107,8 @@ project(':core') {
from(project(':streams:test-utils').configurations.runtimeClasspath) { into("libs/") }
from(project(':streams:examples').jar) { into("libs/") }
from(project(':streams:examples').configurations.runtimeClasspath) { into("libs/") }
from(project(':tools:tools-api').jar) { into("libs/") }

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This causes a resolution issue. I have opened https://issues.apache.org/jira/browse/KAFKA-19750 to fix it

from(project(':tools:tools-api').configurations.runtimeClasspath) { into("libs/") }
duplicatesStrategy 'exclude'
}

Expand Down Expand Up @@ -1766,13 +1769,69 @@ project(':storage') {
}
}

project(':tools:tools-api') {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we change the javadoc part to include the tools.api package?

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The doc-related test is shown in previous comment (#13454 (comment))

archivesBaseName = "kafka-tools-api"

dependencies {
implementation project(':clients')
testImplementation libs.junitJupiter
}

task createVersionFile() {
def receiptFile = file("$buildDir/kafka/$buildVersionFileName")
inputs.property "commitId", commitId
inputs.property "version", version
outputs.file receiptFile

doLast {
def data = [
commitId: commitId,
version: version,
]

receiptFile.parentFile.mkdirs()
def content = data.entrySet().collect { "$it.key=$it.value" }.sort().join("\n")
receiptFile.setText(content, "ISO-8859-1")
}
}

sourceSets {
main {
java {
srcDirs = ["src/main/java"]
}
}
test {
java {
srcDirs = ["src/test/java"]
}
}
}

jar {
dependsOn createVersionFile
from("$buildDir") {
include "kafka/$buildVersionFileName"
}
}

clean.doFirst {
delete "$buildDir/kafka/"
}

javadoc {
include "**/org/apache/kafka/tools/api/*"
}
}

project(':tools') {
archivesBaseName = "kafka-tools"

dependencies {
implementation project(':clients')
implementation project(':server-common')
implementation project(':log4j-appender')
implementation project(':tools:tools-api')
implementation libs.argparse4j
implementation libs.jacksonDatabind
implementation libs.jacksonJDK8Datatypes
Expand Down
1 change: 1 addition & 0 deletions checkstyle/import-control.xml
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@

<subpackage name="common">
<allow class="org.apache.kafka.clients.consumer.ConsumerRecord" exact-match="true" />
<allow class="org.apache.kafka.clients.producer.ProducerRecord" exact-match="true" />
<allow class="org.apache.kafka.common.message.ApiMessageType" exact-match="true" />
<disallow pkg="org.apache.kafka.clients" />
<allow pkg="org.apache.kafka.common" exact-match="true" />
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/common/MessageReader.scala
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import org.apache.kafka.clients.producer.ProducerRecord
* This is used by the `ConsoleProducer`.
*/
@deprecated("This class has been deprecated and will be removed in 4.0." +
"Please use org.apache.kafka.tools.RecordReader instead", "3.5.0")
"Please use org.apache.kafka.tools.api.RecordReader instead", "3.5.0")
trait MessageReader {

def init(inputStream: InputStream, props: Properties): Unit = {}
Expand Down
4 changes: 2 additions & 2 deletions core/src/main/scala/kafka/tools/ConsoleProducer.scala
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ import org.apache.kafka.common.KafkaException
import org.apache.kafka.common.record.CompressionType
import org.apache.kafka.common.utils.Utils
import org.apache.kafka.server.util.{CommandDefaultOptions, CommandLineUtils}
import org.apache.kafka.tools.RecordReader
import org.apache.kafka.tools.api.RecordReader

import scala.annotation.nowarn

Expand All @@ -45,7 +45,7 @@ object ConsoleProducer extends Logging {
r.configure(prop.asInstanceOf[java.util.Map[String, _]])
r
case r: MessageReader =>
logger.warn("MessageReader is deprecated. Please use org.apache.kafka.tools.RecordReader instead")
logger.warn("MessageReader is deprecated. Please use org.apache.kafka.tools.api.RecordReader instead")
new RecordReader {
private[this] var initialized = false

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ import java.nio.file.Files
import kafka.tools.ConsoleProducer.LineMessageReader
import kafka.utils.{Exit, TestUtils}
import org.apache.kafka.clients.producer.{Producer, ProducerConfig, ProducerRecord}
import org.apache.kafka.tools.RecordReader
import org.apache.kafka.tools.api.RecordReader
import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, assertTrue}
import org.junit.jupiter.api.Test
import org.mockito.Mockito
Expand Down
1 change: 1 addition & 0 deletions settings.gradle
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ include 'clients',
'streams:upgrade-system-tests-32',
'streams:upgrade-system-tests-33',
'tools',
'tools:tools-api',
'trogdor'

rootProject.name = 'kafka'
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.tools;
package org.apache.kafka.tools.api;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.Configurable;
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
/*
* 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.
*/
/**
* Provides interfaces for writing plugins of kafka tools
*/
package org.apache.kafka.tools.api;
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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.tools.api;

import org.junit.jupiter.api.Test;

import java.util.Collections;

public class RecordReaderTest {

@Test
void testDefaultCloseAndConfigure() {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do we test in this unit test? There is no assertion.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

make sure the default implementation of those methods don't throw exception.

I will add assertDoesNotThrow to have it more readable.

RecordReader reader = inputStream -> null;
reader.configure(Collections.emptyMap());
reader.close();
}
}