@Test public void testDataChangeEventCanMarshalASpan() { TypeLiteral<DataChangeEvent<ID>> eventType = new TypeLiteral<DataChangeEvent<ID>>() { }; StreamBatchRecord<DataChangeEvent<ID>> sbr = support.lineToEventStreamBatchRecord(spanCtxDataLine, eventType.type(), new LoggingStreamOffsetObserver()); DataChangeEvent<ID> event = sbr.streamBatch().events().get(0); final Map<String, String> spanCtx = event.metadata().spanCtx(); assertEquals(spanCtx.get("ot-tracer-spanid"), "span_01ghst6jv93yj9p8r9ezrh33m1"); assertEquals(spanCtx.get("ot-tracer-traceid"), "trc_01cvst6jv93yj9p8r9ezrh3em0"); assertEquals(spanCtx.get("ot-baggage-foo"), "bar"); }
public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } JsonBatchSupport(JsonSupport jsonSupport); }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } JsonBatchSupport(JsonSupport jsonSupport); StreamBatchRecord<T> lineToEventStreamBatchRecord(
      String line, Type eventType, StreamOffsetObserver observer); StreamBatchRecord<T> lineToSubscriptionStreamBatchRecord(
      String line, Type type, StreamOffsetObserver observer, String xNakadiStreamId,
      String subscriptionId); }
JsonBatchSupport { public <T> StreamBatchRecord<T> lineToEventStreamBatchRecord( String line, Type eventType, StreamOffsetObserver observer) { return new StreamBatchRecordReal<>(marshalEventStreamBatch(line, eventType), observer); } JsonBatchSupport(JsonSupport jsonSupport); StreamBatchRecord<T> lineToEventStreamBatchRecord(
      String line, Type eventType, StreamOffsetObserver observer); StreamBatchRecord<T> lineToSubscriptionStreamBatchRecord(
      String line, Type type, StreamOffsetObserver observer, String xNakadiStreamId,
      String subscriptionId);  }