@Override
  public Iterable<WindowedValue<ElemT>> elementsIterable() {
    return FluentIterable.from(workItem.getMessageBundlesList())
        .transformAndConcat(Windmill.InputMessageBundle::getMessagesList)
        .transform(
            message -> {
              try {
                Instant timestamp =
                    WindmillTimeUtils.windmillToHarnessTimestamp(message.getTimestamp());
                Collection<? extends BoundedWindow> windows =
                    WindmillSink.decodeMetadataWindows(windowsCoder, message.getMetadata());
                PaneInfo pane = WindmillSink.decodeMetadataPane(message.getMetadata());

                InputStream inputStream = message.getData().newInput();
                ElemT value = valueCoder.decode(inputStream, Coder.Context.OUTER);
                return WindowedValue.of(value, timestamp, windows, pane);
              } catch (IOException e) {
                throw new RuntimeException(e);
              }
            });
  }
@Test
  public void testElementIteration() throws Exception {
    Windmill.WorkItem.Builder workItem =
        Windmill.WorkItem.newBuilder().setKey(SERIALIZED_KEY).setWorkToken(17);
    Windmill.InputMessageBundle.Builder chunk1 = workItem.addMessageBundlesBuilder();
    chunk1.setSourceComputationId("computation");
    addElement(chunk1, 5, "hello", WINDOW_1, paneInfo(0));
    addElement(chunk1, 7, "world", WINDOW_2, paneInfo(2));
    Windmill.InputMessageBundle.Builder chunk2 = workItem.addMessageBundlesBuilder();
    chunk2.setSourceComputationId("computation");
    addElement(chunk2, 6, "earth", WINDOW_1, paneInfo(1));

    KeyedWorkItem<String, String> keyedWorkItem =
        new WindmillKeyedWorkItem<>(
            KEY, workItem.build(), WINDOW_CODER, WINDOWS_CODER, VALUE_CODER);

    assertThat(
        keyedWorkItem.elementsIterable(),
        Matchers.contains(
            WindowedValue.of("hello", new Instant(5), WINDOW_1, paneInfo(0)),
            WindowedValue.of("world", new Instant(7), WINDOW_2, paneInfo(2)),
            WindowedValue.of("earth", new Instant(6), WINDOW_1, paneInfo(1))));
  }