public static Destination convertToJcsmpDestination(Solace.Destination destination) {
    if (destination.getType().equals(Solace.DestinationType.TOPIC)) {
      return topicFromName(checkNotNull(destination.getName()));
    } else if (destination.getType().equals(Solace.DestinationType.QUEUE)) {
      return queueFromName(checkNotNull(destination.getName()));
    } else {
      throw new IllegalArgumentException(
          "SolaceIO.Write: Unknown destination type: " + destination.getType());
    }
  }
@Test
  public void testTopicEncoding() {
    MockSessionService mockClientService =
        new MockSessionService(
            index -> {
              List<BytesXMLMessage> messages =
                  ImmutableList.of(
                      SolaceDataUtils.getBytesXmlMessage("payload_test0", "450"),
                      SolaceDataUtils.getBytesXmlMessage("payload_test1", "451"),
                      SolaceDataUtils.getBytesXmlMessage("payload_test2", "452"));
              return getOrNull(index, messages);
            },
            3);

    SessionServiceFactory fakeSessionServiceFactory =
        new MockSessionServiceFactory(mockClientService);

    // Run
    PCollection<Solace.Record> events =
        pipeline.apply(
            "Read from Solace",
            getDefaultRead().withSessionServiceFactory(fakeSessionServiceFactory));

    // Run the pipeline
    PCollection<Boolean> destAreTopics =
        events.apply(
            MapElements.into(TypeDescriptors.booleans())
                .via(
                    r -> {
                      Destination dest = SolaceIO.convertToJcsmpDestination(r.getDestination());
                      return dest instanceof Topic;
                    }));

    List<Boolean> expected = ImmutableList.of(true, true, true);

    // Assert results
    PAssert.that(destAreTopics).containsInAnyOrder(expected);
    pipeline.run();
  }