Skip to content

KAFKA-6955: Use Java AdminClient in DeleteRecordsCommand #5088

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 9 commits into from
Jun 2, 2018
Merged

KAFKA-6955: Use Java AdminClient in DeleteRecordsCommand #5088

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
6e71ca1
KAFKA-6955: Use Java AdminClient in DeleteRecordsCommand
vahidhashemian May 27, 2018
3595e1a
Minor update
vahidhashemian May 30, 2018
b25c8f4
Removed the old method from AdminClient.scala, and updated the unit t…
vahidhashemian Jun 1, 2018
3beab26
Removed the alias for the new AdminClient class
vahidhashemian Jun 1, 2018
ce98717
Moved delete records tests to AdminClientIntegrationTest
vahidhashemian Jun 1, 2018
667939c
A few tweaks
ijuma Jun 2, 2018
b5d6a55
Remove `()`
ijuma Jun 2, 2018
016200d
Fixed the expected exception unit test
vahidhashemian Jun 2, 2018
28d4e05
Minor tweak
ijuma Jun 2, 2018
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
8 changes: 3 additions & 5 deletions core/src/main/scala/kafka/admin/DeleteRecordsCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,11 +73,9 @@ object DeleteRecordsCommand {
out.println("Records delete operation completed:")

deleteRecordsResult.lowWatermarks.asScala.foreach { case (tp, partitionResult) => {
try {
out.println(s"partition: $tp\tlow_watermark: ${partitionResult.get().lowWatermark()}")
} catch {
case e: Exception =>
out.println(s"partition: $tp\terror: ${e.getMessage}")
try out.println(s"partition: $tp\tlow_watermark: ${partitionResult.get.lowWatermark}")
catch {
case e: Exception => out.println(s"partition: $tp\terror: ${e.getMessage}")
}
}}

Expand Down
124 changes: 59 additions & 65 deletions core/src/test/scala/integration/kafka/api/AdminClientIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -742,7 +742,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
assertEquals(0L, consumer.position(topicPartition))

val result = client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(5L)).asJava)
val lowWatermark = result.lowWatermarks().get(topicPartition).get().lowWatermark()
val lowWatermark = result.lowWatermarks().get(topicPartition).get.lowWatermark
assertEquals(5L, lowWatermark)

consumer.seekToBeginning(Collections.singletonList(topicPartition))
Expand All @@ -751,7 +751,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
consumer.seek(topicPartition, 7L)
assertEquals(7L, consumer.position(topicPartition))

client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(DeleteRecordsRequest.HIGH_WATERMARK)).asJava).all().get()
client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(DeleteRecordsRequest.HIGH_WATERMARK)).asJava).all.get
consumer.seekToBeginning(Collections.singletonList(topicPartition))
assertEquals(10L, consumer.position(topicPartition))
}
Expand Down Expand Up @@ -804,7 +804,7 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {

sendRecords(producers.head, 10, topicPartition)
val result = client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(3L)).asJava)
val lowWatermark = result.lowWatermarks().get(topicPartition).get().lowWatermark()
val lowWatermark = result.lowWatermarks.get(topicPartition).get.lowWatermark
assertEquals(3L, lowWatermark)

for (i <- 0 until serverCount)
Expand All @@ -824,14 +824,68 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
assertEquals(0L, consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava).get(topicPartition).offset())

var result = client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(5L)).asJava)
result.all().get()
result.all.get
assertEquals(5L, consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava).get(topicPartition).offset())

result = client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(DeleteRecordsRequest.HIGH_WATERMARK)).asJava)
result.all().get()
result.all.get
assertNull(consumer.offsetsForTimes(Map(topicPartition -> JLong.valueOf(0L)).asJava).get(topicPartition))
}

@Test
def testConsumeAfterDeleteRecords(): Unit = {
val consumer = consumers.head
subscribeAndWaitForAssignment(topic, consumer)

client = AdminClient.create(createConfig)

sendRecords(producers.head, 10, topicPartition)
var messageCount = 0
TestUtils.waitUntilTrue(() => {
messageCount += consumer.poll(0).count()
messageCount == 10
}, "Expected 10 messages", 3000L)

client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(3L)).asJava).all().get()

@vahidhashemian vahidhashemian Jun 2, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

A couple more () can be removed here and a few lines below.

consumer.seek(topicPartition, 1)
messageCount = 0
TestUtils.waitUntilTrue(() => {
messageCount += consumer.poll(0).count()
messageCount == 7
}, "Expected 7 messages", 3000L)

client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(8L)).asJava).all().get()
consumer.seek(topicPartition, 1)
messageCount = 0
TestUtils.waitUntilTrue(() => {
messageCount += consumer.poll(0).count()
messageCount == 2
}, "Expected 2 messages", 3000L)
}

@Test
def testDeleteRecordsWithException(): Unit = {
subscribeAndWaitForAssignment(topic, consumers.head)

client = AdminClient.create(createConfig)

sendRecords(producers.head, 10, topicPartition)

assertEquals(5L, client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(5L)).asJava)
.lowWatermarks.get(topicPartition).get.lowWatermark)

// OffsetOutOfRangeException if offset > high_watermark
intercept[OffsetOutOfRangeException] {
client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(20L)).asJava).lowWatermarks.get(topicPartition).get
}

val nonExistPartition = new TopicPartition(topic, 3)
// LeaderNotAvailableException if non existent partition
intercept[LeaderNotAvailableException] {
client.deleteRecords(Map(nonExistPartition -> RecordsToDelete.beforeOffset(20L)).asJava).lowWatermarks.get(nonExistPartition).get
}
}

@Test
def testDescribeConfigsForTopic(): Unit = {
createTopic(topic, numPartitions = 2, replicationFactor = serverCount)
Expand Down Expand Up @@ -1080,66 +1134,6 @@ class AdminClientIntegrationTest extends IntegrationTestHarness with Logging {
}
}

@Test
def testConsumeAfterDeleteRecords() {
val consumer = consumers.head
subscribeAndWaitForAssignment(topic, consumer)

client = AdminClient.create(createConfig)

sendRecords(producers.head, 10, topicPartition)
var messageCount = 0
TestUtils.waitUntilTrue(() => {
messageCount += consumer.poll(0).count()
messageCount == 10
}, "Expected 10 messages", 3000L)

client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(3L)).asJava).all().get()
consumer.seek(topicPartition, 1)
messageCount = 0
TestUtils.waitUntilTrue(() => {
messageCount += consumer.poll(0).count()
messageCount == 7
}, "Expected 7 messages", 3000L)

client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(8L)).asJava).all().get()
consumer.seek(topicPartition, 1)
messageCount = 0
TestUtils.waitUntilTrue(() => {
messageCount += consumer.poll(0).count()
messageCount == 2
}, "Expected 2 messages", 3000L)
}

@Test
def testDeleteRecordsWithException() {
subscribeAndWaitForAssignment(topic, consumers.head)

client = AdminClient.create(createConfig)

sendRecords(producers.head, 10, topicPartition)
// Should get success result

assertEquals(5L, client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(5L)).asJava)
.lowWatermarks().get(topicPartition).get().lowWatermark())
// OffsetOutOfRangeException if offset > high_watermark
try {
client.deleteRecords(Map(topicPartition -> RecordsToDelete.beforeOffset(20)).asJava).lowWatermarks().get(topicPartition).get()
fail("Expected an offset out of range exception")
} catch {
case e => assertTrue(e.getCause.isInstanceOf[OffsetOutOfRangeException])
}

val nonExistPartition = new TopicPartition(topic, 3)
// UnknownTopicOrPartitionException if user tries to delete records of a non-existent partition
try {
client.deleteRecords(Map(nonExistPartition -> RecordsToDelete.beforeOffset(20)).asJava).lowWatermarks().get(nonExistPartition).get()
fail("Expected a leader not available exception")
} catch {
case e => assertTrue(e.getCause.isInstanceOf[LeaderNotAvailableException])
}
}

}

object AdminClientIntegrationTest {
Expand Down
2 changes: 1 addition & 1 deletion docs/upgrade.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ <h5><a id="upgrade_200_notable" href="#upgrade_200_notable">Notable changes in 2
timeout behavior for blocking APIs. In particular, a new <code>poll(Duration)</code> API has been added which
does not block for dynamic partition assignment. The old <code>poll(long)</code> API has been deprecated and
will be removed in a future version.</li>
<li>The Scala method <code>AdminClient.deleteRecordsBefore</code> has been removed in favor of <code>deleteRecords</code> in the new Java-based AdminClient.</li>
<li>The internal method <code>kafka.admin.AdminClient.deleteRecordsBefore</code> has been removed. Users are encouraged to migrate to <code>org.apache.kafka.clients.admin.AdminClient.deleteRecords</code>.</li>
</ul>

<h5><a id="upgrade_200_new_protocols" href="#upgrade_200_new_protocols">New Protocol Versions</a></h5>
Expand Down