Fixing KafkaInputFormat build failures.

confluentinc · Feb 23, 2022 · 5aa897e · 5aa897e
1 parent 75c15a2
commit 5aa897e
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 6 deletions.
diff --git a/...dexing-service/src/main/java/org/apache/druid/data/input/kafkainput/KafkaInputFormat.java b/...dexing-service/src/main/java/org/apache/druid/data/input/kafkainput/KafkaInputFormat.java
@@ -80,7 +80,7 @@ public boolean isSplittable()
   public InputEntityReader createReader(InputRowSchema inputRowSchema, InputEntity source, File temporaryDirectory)
   {
     KafkaRecordEntity record = (KafkaRecordEntity) source;
-    InputRowSchema newInputRowSchema = new InputRowSchema(dummyTimestampSpec, inputRowSchema.getDimensionsSpec(), inputRowSchema.getMetricNames());
+    InputRowSchema newInputRowSchema = new InputRowSchema(dummyTimestampSpec, inputRowSchema.getDimensionsSpec(), inputRowSchema.getColumnsFilter());
     return new KafkaInputReader(
         inputRowSchema,
         record,

diff --git a/...ng-service/src/test/java/org/apache/druid/data/input/kafkainput/KafkaInputFormatTest.java b/...ng-service/src/test/java/org/apache/druid/data/input/kafkainput/KafkaInputFormatTest.java
@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
+import org.apache.druid.data.input.ColumnsFilter;
 import org.apache.druid.data.input.InputEntityReader;
 import org.apache.druid.data.input.InputRow;
 import org.apache.druid.data.input.InputRowSchema;
@@ -48,7 +49,6 @@
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
-import java.util.Collections;
 
 public class KafkaInputFormatTest
 {
@@ -178,7 +178,7 @@ public void testWithHeaderKeyAndValue() throws IOException
                 "kafka.newheader.kafkapkc",
                 "kafka.newts.timestamp"
             ))),
-            Collections.emptyList()
+            ColumnsFilter.all()
         ),
         inputEntity,
         null
@@ -251,7 +251,7 @@ public void testWithOutKey() throws IOException
                 "kafka.newheader.kafkapkc",
                 "kafka.newts.timestamp"
             ))),
-            Collections.emptyList()
+            ColumnsFilter.all()
         ),
         inputEntity,
         null
@@ -323,7 +323,7 @@ public byte[] value()
                 "kafka.newheader.encoding",
                 "kafka.newheader.kafkapkc"
             ))),
-            Collections.emptyList()
+            ColumnsFilter.all()
         ),
         inputEntity,
         null
@@ -416,7 +416,7 @@ public void testWithOutKeyAndHeaderSpecs() throws IOException
                 "bar", "foo",
                 "kafka.newts.timestamp"
             ))),
-            Collections.emptyList()
+            ColumnsFilter.all()
         ),
         inputEntity,
         null