[CONNECTOR] Fix some issues in connector

common/src/main/java/org/astraea/common/Configuration.java

@@ -103,17 +103,17 @@ default String requireString(String key) {
   }
 
   /**
-   * @param filteredString the string to be filtered and removed
-   * @return new Configuration only contains which the key value includes the filteredString, and
-   *     the filteredString and the following dot will be removed from the key
+   * @param prefix the string to be filtered and removed
+   * @return new Configuration only contains which the key value starts with the prefix, and the
+   *     prefix string and the following dot will be removed from the key
    */
-  default Configuration filteredConfigs(String filteredString) {
+  default Configuration filteredConfigs(String prefix) {
     return of(
         entrySet().stream()
-            .filter(k -> k.getKey().contains(filteredString))
+            .filter(k -> k.getKey().startsWith(prefix))
             .collect(
                 Collectors.toMap(
-                    i -> i.getKey().replaceAll(filteredString + '.', ""), Map.Entry::getValue)));
+                    i -> i.getKey().replaceFirst(prefix + '.', ""), Map.Entry::getValue)));
   }
 
   /**

common/src/test/java/org/astraea/common/partitioner/ConfigurationTest.java

@@ -47,7 +47,7 @@ void testMap() {
 
   @Test
   void testFilteredConfigs() {
-    var config = Configuration.of(Map.of("key", "v1", "filtered.key", "v2"));
+    var config = Configuration.of(Map.of("key", "v1", "filtered.key", "v2", "key.filtered", "v3"));
     Assertions.assertEquals(Map.of("key", "v2"), config.filteredConfigs("filtered").raw());
   }
 }