@@ -113,10 +113,25 @@ public class KinesisSinkConfig extends BaseKinesisConfig implements Serializable
113113 defaultValue = "" ,
114114 help = "Path to the native Amazon Kinesis Producer Library (KPL) binary.\n "
115115 + "Only use this setting if you want to use a custom build of the native code.\n "
116- + "This setting can also be set with the environment variable `PULSAR_IO_KINESIS_KPL_PATH`.\n "
116+ + "This setting can also be set with the environment variable `PULSAR_IO_KINESIS_KPL_1_0_PATH`"
117+ + "or `PULSAR_IO_KINESIS_KPL_PATH`.\n "
117118 + "If not set, the Kinesis sink will use the built-in native executable."
118119 )
119- private String nativeExecutable = System .getenv ("PULSAR_IO_KINESIS_KPL_PATH" );
120+ private String nativeExecutable = resolveDefaultKinesisProducerLibraryPath ();
121+
122+ private static String resolveDefaultKinesisProducerLibraryPath () {
123+ // Prefer PULSAR_IO_KINESIS_KPL_1_0_PATH environment variable over PULSAR_IO_KINESIS_KPL_PATH.
124+ // This setting supports building a Pulsar Functions base image that is used to run different Pulsar IO Kinesis
125+ // sink versions. The older versions of Pulsar IO Kinesis sink can continue to use the binary configured with
126+ // PULSAR_IO_KINESIS_KPL_PATH, pointing to a 0.15.12 native executable. The newer versions of Pulsar IO Kinesis
127+ // sink can use the binary configured with PULSAR_IO_KINESIS_KPL_1_0_PATH, pointing to a 1.0.4
128+ // native executable.
129+ String kplPath = System .getenv ("PULSAR_IO_KINESIS_KPL_1_0_PATH" );
130+ if (isNotBlank (kplPath )) {
131+ return kplPath ;
132+ }
133+ return System .getenv ("PULSAR_IO_KINESIS_KPL_PATH" );
134+ }
120135
121136 public static KinesisSinkConfig load (Map <String , Object > config , SinkContext sinkContext ) {
122137 KinesisSinkConfig kinesisSinkConfig = IOConfigUtils .loadWithSecrets (config , KinesisSinkConfig .class , sinkContext );
0 commit comments