INNER CODE UNIT · Java

getSourceProperties

mongodb/mongo-kafka · src/integrationTest/java/com/mongodb/kafka/connect/ChangeStreamRoundTripTest.java:282

  private Properties getSourceProperties(final MongoDatabase database) {
    Properties sourceProperties = new Properties();
    sourceProperties.put(DATABASE_CONFIG, database.getName());
    sourceProperties.put(TOPIC_PREFIX_CONFIG, "copy");
    sourceProperties.put(STARTUP_MODE_CONFIG, StartupMode.COPY_EXISTING.propertyValue());
    return sourceProperties;
  }

  @NotNull
  private Properties getSinkProperties(
      final MongoDatabase destination, final MongoCollection<?>... sources) {
    Properties sinkProperties = createSinkProperties();

    String topics =
        Arrays.stream(sources)
            .map(coll -> format("copy.%s", coll.getNamespace().getFullName()))
            .collect(Collectors.joining(", "));
    sinkProperties.put("topics", topics);

View source record →