Using Admin API

Properties p = new Properties();
p.put("bootstrap.servers", "localhost:9092");
Admin admin = Admin.create(p);

admin.createTopics(List.of(
  new NewTopic("orders", 6, (short) 3))).all().get();

Set<String> topics = admin.listTopics().names().get();

TopicDescription d = admin.describeTopics(List.of("orders"))
  .allTopicNames().get().get("orders");
System.out.println(d.partitions().size());

admin.deleteTopics(List.of("obsolete-topic")).all().get();

admin.createPartitions(
  Map.of("orders", NewPartitions.increaseTo(12))).all().get();

ConfigResource cr = new ConfigResource(ConfigResource.Type.TOPIC, "orders");
admin.incrementalAlterConfigs(Map.of(cr, List.of(
  new AlterConfigOp(new ConfigEntry("retention.ms","604800000"),
    AlterConfigOp.OpType.SET)))).all().get();

Config c = admin.describeConfigs(List.of(cr)).all().get().get(cr);
c.entries().forEach(e -> System.out.println(e.name()+"="+e.value()));

ConsumerGroupDescription g = admin
  .describeConsumerGroups(List.of("order-svc"))
  .all().get().get("order-svc");

admin.listConsumerGroups().valid().get()
  .forEach(g -> System.out.println(g.groupId()));

admin.deleteConsumerGroups(List.of("old-group")).all().get();

TopicPartition tp = new TopicPartition("orders", 0);
admin.deleteRecords(Map.of(tp, RecordsToDelete.beforeOffset(1000)))
  .all().get();