Add support for dynamic write in MqttIO

[twosom]

Please add a meaningful description for your change here

fixes #19376

This PR contains these changes

• add support for dynamic write in MqttIO
• add related test

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Assigning reviewers. If you would like to opt out of this review, comment assign to next reviewer:

R: @kennknowles for label java.
R: @damondouglas for label io.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

Thanks for the contribution. The added PTransform and DoFn pretty much duplicated the existing one. I would suggest adjust the current Write transform and WriteFn for this feature. This avoids branches and more independent transform/DoFn to maintain. Should be straightforward to do so.

I see, there was concern of breaking change, Write originally is not a generic but WriteDynamics transform is a generic. We can make Write to return a WriteDynamics<byte[]> with fixed topicFn and identity payloadFn.

[twosom]

Thanks for the contribution. The added PTransform and DoFn pretty much duplicated the existing one. I would suggest adjust the current Write transform and WriteFn for this feature. This avoids branches and more independent transform/DoFn to maintain. Should be straightforward to do so.

I see, there was concern of breaking change, Write originally is not a generic but WriteDynamics transform is a generic. We can make Write to return a WriteDynamics<byte[]> with fixed topicFn and identity payloadFn.

@Abacn
Thanks for your comment. I'll make the improvements you suggested.

[Abacn]

Hi, I do not see code update. Was the commit pushed to origin?

[twosom]

Hi, I do not see code update. Was the commit pushed to origin?

Nope, I just merged conflicted ^_^

and I'm doing working this MqttIO

[twosom]

@Abacn
and I updated code just now.

Thanks for your comment. ^_^

[Abacn]

since getTopic() is or not null corresponds to dynamic() being false and true, we can remove this property, and just use topic nullness to determine the scenario?

[twosom]

This would also allow calls like MqttIO.write().withTopicPayloadFn(...), since they use the same Write class.
We don't know if the Write object was called from write() or dynamicWrite(). This is not enough to determine the scenario based on whether the topic is null.
The topic value can be modified externally, but the dynamic value cannot.
I would like to make sure that withTopicPayloadFn() and withPayloadFn() are only available when dynamic is true. What do you think? and please take a look a the following test code.

@Test
  public void testWriteWithTopicFn() {
    IllegalArgumentException exception =
        assertThrows(
            IllegalArgumentException.class, () -> MqttIO.write().withTopicFn(e -> "some topic"));

    assertEquals("withTopicFn can not use in non-dynamic write", exception.getMessage());
  }

  @Test
  public void testWriteWithPayloadFn() {
    final IllegalArgumentException exception =
        assertThrows(
            IllegalArgumentException.class, () -> MqttIO.write().withPayloadFn(e -> new byte[] {}));

    assertEquals("withPayloadFn can not use in non-dynamic write", exception.getMessage());
  }