KAFKA-6955: Use Java AdminClient in DeleteRecordsCommand

core/src/main/scala/kafka/admin/AdminClient.scala

@@ -18,7 +18,6 @@ import java.util.{Collections, Properties}
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.{ConcurrentLinkedQueue, Future, TimeUnit}
 
-import kafka.admin.AdminClient.DeleteRecordsResult
 import kafka.common.KafkaException
 import kafka.coordinator.group.GroupOverview
 import kafka.utils.Logging
@@ -216,73 +215,6 @@ class AdminClient(val time: Time,
       broker -> Try[NodeApiVersions](new NodeApiVersions(getApiVersions(broker).asJava))
     }.toMap
 
-  /*
-   * Remove all the messages whose offset is smaller than the given offset of the corresponding partition
-   *
-   * DeleteRecordsResult contains either lowWatermark of the partition or exception. We list the possible exception
-   * and their interpretations below:
-   *
-   * - DisconnectException if leader node of the partition is not available. Need retry by user.
-   * - PolicyViolationException if the topic is configured as non-deletable.
-   * - TopicAuthorizationException if the topic doesn't exist and the user doesn't have the authority to create the topic
-   * - TimeoutException if response is not available within the timeout specified by either Future's timeout or AdminClient's request timeout
-   * - UnknownTopicOrPartitionException if the partition doesn't exist or if the user doesn't have the authority to describe the topic
-   * - NotLeaderForPartitionException if broker is not leader of the partition. Need retry by user.
-   * - OffsetOutOfRangeException if the offset is larger than high watermark of this partition
-   *
-   */
-
-  def deleteRecordsBefore(offsets: Map[TopicPartition, Long]): Future[Map[TopicPartition, DeleteRecordsResult]] = {
-    val metadataRequest = new MetadataRequest.Builder(offsets.keys.map(_.topic).toSet.toList.asJava, true)
-    val response = sendAnyNode(ApiKeys.METADATA, metadataRequest).asInstanceOf[MetadataResponse]
-    val errors = response.errors
-    if (!errors.isEmpty)
-      error(s"Metadata request contained errors: $errors")
-
-    val (partitionsWithoutError, partitionsWithError) = offsets.partition{ partitionAndOffset =>
-      !response.errors().containsKey(partitionAndOffset._1.topic())}
-
-    val (partitionsWithLeader, partitionsWithoutLeader) = partitionsWithoutError.partition{ partitionAndOffset =>
-      response.cluster().leaderFor(partitionAndOffset._1) != null}
-
-    val partitionsWithErrorResults = partitionsWithError.keys.map( partition =>
-      partition -> DeleteRecordsResult(DeleteRecordsResponse.INVALID_LOW_WATERMARK, response.errors().get(partition.topic()).exception())).toMap
-
-    val partitionsWithoutLeaderResults = partitionsWithoutLeader.mapValues( _ =>
-      DeleteRecordsResult(DeleteRecordsResponse.INVALID_LOW_WATERMARK, Errors.LEADER_NOT_AVAILABLE.exception()))
-
-    val partitionsGroupByLeader = partitionsWithLeader.groupBy(partitionAndOffset =>
-      response.cluster().leaderFor(partitionAndOffset._1))
-
-    // prepare requests and generate Future objects
-    val futures = partitionsGroupByLeader.map{ case (node, partitionAndOffsets) =>
-      val convertedMap: java.util.Map[TopicPartition, java.lang.Long] = partitionAndOffsets.mapValues(_.asInstanceOf[java.lang.Long]).asJava
-      val future = client.send(node, new DeleteRecordsRequest.Builder(requestTimeoutMs, convertedMap))
-      pendingFutures.add(future)
-      future.compose(new RequestFutureAdapter[ClientResponse, Map[TopicPartition, DeleteRecordsResult]]() {
-          override def onSuccess(response: ClientResponse, future: RequestFuture[Map[TopicPartition, DeleteRecordsResult]]) {
-            val deleteRecordsResponse = response.responseBody().asInstanceOf[DeleteRecordsResponse]
-            val result = deleteRecordsResponse.responses().asScala.mapValues(v => DeleteRecordsResult(v.lowWatermark, v.error.exception())).toMap
-            future.complete(result)
-            pendingFutures.remove(future)
-          }
-
-          override def onFailure(e: RuntimeException, future: RequestFuture[Map[TopicPartition, DeleteRecordsResult]]) {
-            val result = partitionAndOffsets.mapValues(_ => DeleteRecordsResult(DeleteRecordsResponse.INVALID_LOW_WATERMARK, e))
-            future.complete(result)
-            pendingFutures.remove(future)
-          }
-
-        })
-    }
-
-    // default output if not receiving DeleteRecordsResponse before timeout
-    val defaultResults = offsets.mapValues(_ =>
-      DeleteRecordsResult(DeleteRecordsResponse.INVALID_LOW_WATERMARK, Errors.REQUEST_TIMED_OUT.exception())) ++ partitionsWithErrorResults ++ partitionsWithoutLeaderResults
-
-    new CompositeFuture(time, defaultResults, futures.toList)
-  }
-
   /**
    * Case class used to represent a consumer of a consumer group
    */
@@ -473,8 +405,6 @@ object AdminClient {
     config
   }
 
-  case class DeleteRecordsResult(lowWatermark: Long, error: Exception)
-
   class AdminConfig(originals: Map[_,_]) extends AbstractConfig(AdminConfigDef, originals.asJava, false)
 
   def createSimplePlaintext(brokerUrl: String): AdminClient = {

core/src/main/scala/kafka/admin/DeleteRecordsCommand.scala

@@ -20,14 +20,17 @@ package kafka.admin
 import java.io.PrintStream
 import java.util.Properties
 
-import kafka.admin.AdminClient.DeleteRecordsResult
 import kafka.common.AdminCommandFailedException
-import kafka.utils.{CoreUtils, Json, CommandLineUtils}
+import kafka.utils.{CommandLineUtils, CoreUtils, Json}
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.clients.admin
+import org.apache.kafka.clients.admin.RecordsToDelete
 import org.apache.kafka.clients.CommonClientConfigs
 import joptsimple._
 
+import scala.collection.JavaConverters._
+
 /**
  * A command for delete records of the given partitions down to the specified offset.
  */
@@ -61,26 +64,33 @@ object DeleteRecordsCommand {
     if (duplicatePartitions.nonEmpty)
       throw new AdminCommandFailedException("Offset json file contains duplicate topic partitions: %s".format(duplicatePartitions.mkString(",")))
 
+    val recordsToDelete = offsetSeq.map { case (topicPartition, offset) =>
+      (topicPartition, RecordsToDelete.beforeOffset(offset))
+    }.toMap.asJava
+
     out.println("Executing records delete operation")
-    val deleteRecordsResult: Map[TopicPartition, DeleteRecordsResult] = adminClient.deleteRecordsBefore(offsetSeq.toMap).get()
+    val deleteRecordsResult = adminClient.deleteRecords(recordsToDelete)
     out.println("Records delete operation completed:")
 
-    deleteRecordsResult.foreach{ case (tp, partitionResult) => {
-      if (partitionResult.error == null)
-        out.println(s"partition: $tp\tlow_watermark: ${partitionResult.lowWatermark}")
-      else
-        out.println(s"partition: $tp\terror: ${partitionResult.error.toString}")
+    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}")
+      }
     }}
+
     adminClient.close()
   }
 
-  private def createAdminClient(opts: DeleteRecordsCommandOptions): AdminClient = {
+  private def createAdminClient(opts: DeleteRecordsCommandOptions): admin.AdminClient = {
     val props = if (opts.options.has(opts.commandConfigOpt))
       Utils.loadProps(opts.options.valueOf(opts.commandConfigOpt))
     else
       new Properties()
     props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, opts.options.valueOf(opts.bootstrapServerOpt))
-    AdminClient.create(props)
+    admin.AdminClient.create(props)
   }
 
   class DeleteRecordsCommandOptions(args: Array[String]) {

core/src/test/scala/integration/kafka/api/LegacyAdminClientTest.scala

@@ -16,21 +16,24 @@
  */
 package kafka.api
 
-import java.util.Collections
-import java.util.concurrent.TimeUnit
+import java.util.{Collections, Properties}
 
 import kafka.admin.AdminClient
-import kafka.admin.AdminClient.DeleteRecordsResult
 import kafka.server.KafkaConfig
 import java.lang.{Long => JLong}
+import java.util.concurrent.TimeUnit
+
 import kafka.utils.{Logging, TestUtils}
-import org.apache.kafka.clients.consumer.{KafkaConsumer, ConsumerConfig}
-import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, ProducerConfig}
+import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
+import org.apache.kafka.clients.admin.{RecordsToDelete, AdminClient => JAdminClient}
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
 import org.apache.kafka.common.TopicPartition
-import org.apache.kafka.common.protocol.{Errors, ApiKeys}
+import org.apache.kafka.common.errors.{LeaderNotAvailableException, OffsetOutOfRangeException}
+import org.apache.kafka.common.protocol.ApiKeys
 import org.apache.kafka.common.requests.DeleteRecordsRequest
 import org.junit.{After, Before, Test}
 import org.junit.Assert._
+
 import scala.collection.JavaConverters._
 
 /**
@@ -51,6 +54,7 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
   val tp2 = new TopicPartition(topic, part2)
 
   var client: AdminClient = null
+  var jClient : JAdminClient = null
 
   // configure the servers and clients
   this.serverConfig.setProperty(KafkaConfig.ControlledShutdownEnableProp, "false") // speed up shutdown
@@ -69,12 +73,18 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
   override def setUp() {
     super.setUp()
     client = AdminClient.createSimplePlaintext(this.brokerList)
+
+    val properties = new Properties
+    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerList)
+    jClient = JAdminClient.create(properties)
+
     createTopic(topic, 2, serverCount)
   }
 
   @After
   override def tearDown() {
     client.close()
+    jClient.close()
     super.tearDown()
   }
 
@@ -87,11 +97,11 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
     consumer.seekToBeginning(Collections.singletonList(tp))
     assertEquals(0L, consumer.position(tp))
 
-    client.deleteRecordsBefore(Map((tp, 5L))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(5L)).asJava).all().get()
     consumer.seekToBeginning(Collections.singletonList(tp))
     assertEquals(5L, consumer.position(tp))
 
-    client.deleteRecordsBefore(Map((tp, DeleteRecordsRequest.HIGH_WATERMARK))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(DeleteRecordsRequest.HIGH_WATERMARK)).asJava).all().get()
     consumer.seekToBeginning(Collections.singletonList(tp))
     assertEquals(10L, consumer.position(tp))
   }
@@ -108,15 +118,15 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
       messageCount == 10
     }, "Expected 10 messages", 3000L)
 
-    client.deleteRecordsBefore(Map((tp, 3L))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(3L)).asJava).all().get()
     consumer.seek(tp, 1)
     messageCount = 0
     TestUtils.waitUntilTrue(() => {
       messageCount += consumer.poll(0).count()
       messageCount == 7
     }, "Expected 7 messages", 3000L)
 
-    client.deleteRecordsBefore(Map((tp, 8L))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(8L)).asJava).all().get()
     consumer.seek(tp, 1)
     messageCount = 0
     TestUtils.waitUntilTrue(() => {
@@ -130,19 +140,22 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
     subscribeAndWaitForAssignment(topic, consumers.head)
 
     sendRecords(producers.head, 10, tp)
-    assertEquals(DeleteRecordsResult(5L, null), client.deleteRecordsBefore(Map((tp, 5L))).get()(tp))
+    assertEquals(5L, jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(5L)).asJava).lowWatermarks().get(tp).get().lowWatermark())
 
     for (i <- 0 until serverCount)
       killBroker(i)
     restartDeadBrokers()
 
-    client.close()
+    jClient.close()
     brokerList = TestUtils.bootstrapServers(servers, listenerName)
-    client = AdminClient.createSimplePlaintext(brokerList)
+    val properties = new Properties
+    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.brokerList)
+    jClient = JAdminClient.create(properties)
 
     TestUtils.waitUntilTrue(() => {
       // Need to retry if leader is not available for the partition
-      client.deleteRecordsBefore(Map((tp, 0L))).get(1000L, TimeUnit.MILLISECONDS)(tp).equals(DeleteRecordsResult(5L, null))
+      jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(0L)).asJava).lowWatermarks().get(tp)
+        .get(1000L, TimeUnit.MILLISECONDS).lowWatermark().equals(5L)
     }, "Expected low watermark of the partition to be 5L")
   }
 
@@ -151,7 +164,7 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
     subscribeAndWaitForAssignment(topic, consumers.head)
 
     sendRecords(producers.head, 10, tp)
-    client.deleteRecordsBefore(Map((tp, 3L))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(3L)).asJava).all().get()
 
     for (i <- 0 until serverCount)
       assertEquals(3, servers(i).replicaManager.getReplica(tp).get.logStartOffset)
@@ -171,10 +184,10 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
     sendRecords(producers.head, 10, tp)
     assertEquals(0L, consumer.offsetsForTimes(Map(tp -> JLong.valueOf(0L)).asJava).get(tp).offset())
 
-    client.deleteRecordsBefore(Map((tp, 5L))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(5L)).asJava).all().get()
     assertEquals(5L, consumer.offsetsForTimes(Map(tp -> JLong.valueOf(0L)).asJava).get(tp).offset())
 
-    client.deleteRecordsBefore(Map((tp, DeleteRecordsRequest.HIGH_WATERMARK))).get()
+    jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(DeleteRecordsRequest.HIGH_WATERMARK)).asJava).all().get()
     assertNull(consumer.offsetsForTimes(Map(tp -> JLong.valueOf(0L)).asJava).get(tp))
   }
 
@@ -184,14 +197,24 @@ class LegacyAdminClientTest extends IntegrationTestHarness with Logging {
 
     sendRecords(producers.head, 10, tp)
     // Should get success result
-    assertEquals(DeleteRecordsResult(5L, null), client.deleteRecordsBefore(Map((tp, 5L))).get()(tp))
+
+    assertEquals(5L, jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(5L)).asJava).lowWatermarks().get(tp).get().lowWatermark())
     // OffsetOutOfRangeException if offset > high_watermark
-    assertEquals(DeleteRecordsResult(-1L, Errors.OFFSET_OUT_OF_RANGE.exception()), client.deleteRecordsBefore(Map((tp, 20))).get()(tp))
+    try {
+      jClient.deleteRecords(Map(tp -> RecordsToDelete.beforeOffset(20)).asJava).lowWatermarks().get(tp).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
-    assertEquals(DeleteRecordsResult(-1L, Errors.LEADER_NOT_AVAILABLE.exception()),
-                 client.deleteRecordsBefore(Map((nonExistPartition, 20))).get()(nonExistPartition))
+    try {
+      jClient.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])
+    }
   }
 
   @Test

docs/upgrade.html

@@ -95,6 +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 implementation of DeleteRecordsCommand makes use of the new Java-based AdminClient.</li>
 </ul>
 
 <h5><a id="upgrade_200_new_protocols" href="#upgrade_200_new_protocols">New Protocol Versions</a></h5>