Remove prefixes

These can be accomplished entirely in configuration.

src/main/scala/fable/Config.scala

@@ -48,21 +48,16 @@ object Config {
     * General configuration options for [[Kafka]] instances.
     *
     * @constructor
-    * @param prefix optional prefix to apply to topic names and consumer group
-    * IDs.
     * @param uris bootstrap URIs used to connect to a Kafka cluster.
     */
-  case class Kafka(prefix: Option[String], uris: URIList) {
+  case class Kafka(uris: URIList) {
     private[fable] def properties: Properties = {
       val result = new Properties()
       result.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
                  uris.bootstrapServers)
       result.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, uris.scheme)
       result
     }
-
-    def prefix[A](name: String)(f: String => A): A =
-      f(s"${prefix.getOrElse("")}$name")
   }
 
   object Kafka {

src/main/scala/fable/Kafka.scala

@@ -38,20 +38,6 @@ import org.apache.kafka.clients.consumer.KafkaConsumer
   */
 class Kafka[F[_]: ContextShift: Monad: Sync](config: Config.Kafka) {
 
-  /**
-    * Construct a [[Topic]] from the given name. Applies the prefix from
-    * configuration if one is present.
-    */
-  def topic(name: String): Topic =
-    config.prefix(name)(Topic.apply)
-
-  /**
-    * Construct a [[GroupId]] from the given name. Applies the prefix from
-    * configuration if one is present.
-    */
-  def groupId(name: String): GroupId =
-    config.prefix(name)(GroupId.apply)
-
   /**
     * Allocate a [[Consumer]] as a [[cats.effect.Resource]]. The consumer will
     * be closed when the resource is released.

src/test/resources/logback.xml

@@ -2,7 +2,7 @@
 <configuration>
     <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
         <encoder>
-            <pattern>[%level] [%logger{0}] %msg%n</pattern>
+            <pattern>[%level] [%logger] %msg%n</pattern>
         </encoder>
     </appender>
 

src/test/scala/fable/ConsumerSpec.scala

@@ -14,7 +14,7 @@ import scala.concurrent.ExecutionContext
 
 class ConsumerSpec extends AsyncFunSuite {
   test("poll") {
-    val topic = kafka.topic("fable-test-example")
+    val topic = Topic("fable-test-example")
     val consumer = kafka.consumer[String, String](consumerConfig)
 
     (for {
@@ -35,7 +35,7 @@ class ConsumerSpec extends AsyncFunSuite {
   }
 
   test("commit") {
-    val topic = kafka.topic("fable-test-example")
+    val topic = Topic("fable-test-example")
     val consumer = Eval.always {
       kafka.consumer[String, String](consumerConfig)
     }
@@ -66,7 +66,7 @@ class ConsumerSpec extends AsyncFunSuite {
   }
 
   test("records") {
-    val topic = kafka.topic("fable-test-example")
+    val topic = Topic("fable-test-example")
     val consumer =
       kafka.consumer[String, String](consumerConfig.copy(maxPollRecords = 2))
 
@@ -90,7 +90,7 @@ class ConsumerSpec extends AsyncFunSuite {
   }
 
   test("partitionsFor") {
-    val topic = kafka.topic("fable-test-example")
+    val topic = Topic("fable-test-example")
     val consumer = kafka.consumer[String, String](consumerConfig)
 
     (for {
@@ -102,8 +102,8 @@ class ConsumerSpec extends AsyncFunSuite {
   }
 
   test("assign") {
-    val first = kafka.topic("fable-test-one")
-    val second = kafka.topic("fable-test-two")
+    val first = Topic("fable-test-one")
+    val second = Topic("fable-test-two")
     val consumer = kafka.consumer[String, String](consumerConfig)
 
     (for {