org.apache.kafka.connect.errors.DataException: структурные схемы не совпадают

Question

Попытка создать и заполнить схему массива типа string внутри схемы массива определенного типа объекта, используя kafka.connect.data.Schema в java. Но получаю ошибку

org.apache.kafka.connect.errors.DataException: Структурные схемы не совпадают. в org.apache.kafka.connect.data.ConnectSchema.validateValue (ConnectSchema.java:239) в org.apache.kafka.connect.data.ConnectSchema.validateValue (ConnectSchema.java:245) в org.apache.kafka.connect.data.Struct.put (Struct.java:215) в org.apache.kafka.connect.data.Struct.put (Struct.java:204) в com.sintec.ra.fixture.BookProgramFixture.createAiringStruct (BookProgramFixture.java:60) в com.sintec.ra.handlers.BookProgramTopicHandlerTest.mockAiringStruct (BookProgramTopicHandlerTest.java:132) в com.sintec.ra.handlers.BookProgramTopicHandlerTest.init (BookProgramTopicHandlerTest.java:107) в com.sintec.ra.handlers.BookProgramTopicHandlerTest. (BookProgramTopicHandlerTest.java:76) at sun.reflect.NativeConstructorAccessorImpl.newInstance0 (собственный метод) at sun.reflect.NativeConstructorAccessorImpl.newInstance (NativeConstructorAccessorImpl.java:62) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance (DelegatingConstructorAccessorImpl.java:45) в java.lang.reflect.Constructor.newInstance (Constructor.java:423) в org.junit.runners.BlockJUnit4ClassRunner.createTest (BlockJUnit4ClassRunner.java:217) в org.junit.runners.BlockJUnit4ClassRunner $ 1.runReflectiveCall (BlockJUnit4ClassRunner.java:266) в org.junit.internal.runners.model.ReflectiveCallable.run (ReflectiveCallable.java:12) в org.junit.runners.BlockJUnit4ClassRunner.methodBlock (BlockJUnit4ClassRunner.java:263) в org.junit.runners.BlockJUnit4ClassRunner.runChild (BlockJUnit4ClassRunner.java:78) в org.junit.runners.BlockJUnit4ClassRunner.runChild (BlockJUnit4ClassRunner.java:57) в org.junit.runners.ParentRunner $ 3.run (ParentRunner.java:290) в org.junit.runners.ParentRunner $ 1.schedule (ParentRunner.java:71) в org.junit.runners.ParentRunner.runChildren (ParentRunner.java:288) в org.junit.runners.ParentRunner.access $ 000 (ParentRunner.java:58) в org.junit.runners.ParentRunner $ 2.evaluate (ParentRunner.java:268) в org.junit.runners.ParentRunner.run (ParentRunner.java:363) в org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run (JUnit4TestReference.java:86) в org.eclipse.jdt.internal.junit.runner.TestExecution.run (TestExecution.java:38) в org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests (RemoteTestRunner.java:538) в org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests (RemoteTestRunner.java:760) в org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run (RemoteTestRunner.java:460) в org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main (RemoteTestRunner.java:206) Например,

"airingTo" : [ {
      "dow" : [ "SATURDAY" ]
      }]

---

public static Struct createAiringStruct() 
{
    Schema valueSchema = createValueSchemaForProgram();
    Struct valueStruct = new Struct(valueSchema).put("airingTo", 
    getAiringList());
    return valueStruct;
}

private static Schema createValueSchemaForProgram() 
{
    return 
    SchemaBuilder.struct().name("PROGRAM").field("airingTo", 
    SchemaBuilder.array(createAiringListSchema())).build();
}

private static Schema createAiringListSchema() 
{
   return SchemaBuilder.struct().name("airingTo").field("dow", 
   SchemaBuilder.array(Schema.STRING_SCHEMA)).build();
}

private static List<Struct> getAiringList() 
{
    Struct valueStruct = new 
    Struct(createAiringListSchema()).put("dow", 
    Arrays.asList("SATURDAY"));
    List<Struct> dayPartList = new ArrayList<>();
    dayPartList.add(valueStruct);
    return dayPartList;
}

Answer 1

Не знаю, откуда у вас .name("PROGRAM"), но мне кажется, это работает.

Кажется, вы смешивали имена структур с полями.

    Schema dowSchema = SchemaBuilder.array(Schema.STRING_SCHEMA).build();
    Schema airingToItemSchema = SchemaBuilder.struct()
            .field("dow", dowSchema)
            .build();
    Schema airingToSchema = SchemaBuilder.array(airingToItemSchema).build();
    Schema rootSchema = SchemaBuilder.struct()
            .field("airingTo", airingToSchema);
            .build();

    Struct item = new Struct(airingToItemSchema)
            .put("dow", Collections.singletonList("SATURDAY"));
    Struct rootStruct = new Struct(rootSchema)
            .put("airingTo", Collections.singletonList(item));

    System.out.println(rootStruct);

Выходы

Struct{airingTo=[Struct{dow=[SATURDAY]}]}